Construir un pipeline simple GraphRAG con Neo4j, Gemini y LangChain
Un pequeño pipeline GraphRAG con Neo4j, Gemini y LangChain: extracción restringida por esquema, consultas de vecindad y de múltiples saltos, conversión de texto a Cypher de solo lectura.
Fuentes: Building a Simple GraphRAG Pipeline with Neo4j, Gemini, and LangChain
Este tutorial crea una pequeña pila GraphRAG: Neo4j almacena el grafo, Gemini proporciona el modelo de lenguaje y LangChain conecta las partes. El dominio de ejemplo son notas sobre errores en la cocina de un cocinero, elegidos intencionadamente como algo común, para que pueda observar el flujo de trabajo en lugar de tener que entender jerga especializada.
Configuración
Instale el controlador de Neo4j además de los paquetes LangChain Neo4j y Google GenAI:
pip install neo4j langchain-neo4j langchain-google-genai
Conexión a Neo4j
Obtenga la URI, el usuario, la contraseña y, opcionalmente, el nombre de la base de datos desde un archivo local de credenciales para evitar guardar información confidencial:
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")
Haga pruebas preliminares con el controlador de Neo4j estándar antes de incorporar LangChain. Esto permite aislar problemas de inicio de sesión y de red:
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.")
Una vez que el controlador responda, utilice las mismas credenciales con la función auxiliar Neo4jGraph de LangChain que se empleará en los pasos siguientes:
from langchain_neo4j import Neo4jGraph
graph = Neo4jGraph(
url=NEO4J_URI,
username=NEO4J_USERNAME,
password=NEO4J_PASSWORD,
database=NEO4J_DATABASE,
)
Inicializar Gemini
Cree un cliente ChatGoogleGenerativeAI a partir de una clave API del entorno:
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)
Una instancia del modelo cumple dos funciones: transformar informes en bruto en documentos de grafo y convertir preguntas en inglés a lenguaje Cypher.
Construir un grafo a partir de unos pocos documentos
Cuatro archivos cortos que describen incidentes representan un registro de errores. Cárguelos como objetos 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()
]
Controlar estrictamente la extracción de información. Defina ALLOWED_NODES y ALLOWED_RELATIONSHIPS como triples estructurados con los nombres from-label, edge y to-label, de modo que los informes compartan un mismo vocabulario en lugar de crear verbos nuevos para hechos idénticos.
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")
Una ejecución típica podría mostrar conteos como estos:
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
Escriba los documentos del grafo en Neo4j y actualice la metadatos del esquema:
graph.add_graph_documents(graph_documents, baseEntityLabel=True, include_source=True)
graph.refresh_schema()
Consultas directas que muestran el valor estructural
No se requiere una segunda llamada al modelo para estas verificaciones: los bordes ya existen.
Inventariee lo que produjo la extracción:
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"
)
Vecindad a dos pasos de una entidad. La búsqueda por similitud no puede reemplazar a “todo lo conectado con 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},
)
Ruta más corta entre personas que nunca aparecen juntas en un mismo informe. Meera y Devi nunca se mencionan entre sí, pero ambos incidentes comparten una categoría de causa normalizada:
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
Esa ruta es una respuesta verificable sobre la relación entre elementos: primero la estructura, luego el texto descriptivo.
Haga preguntas en inglés sencillo
GraphCypherQAChain es el clásico ciclo de texto a Cypher: el modelo elabora el código Cypher a partir del esquema, Neo4j lo ejecuta y una segunda pasada describe las filas. Dado que el Cypher generado se aplica a una base de datos en tiempo real, añada una instrucción explícita de solo lectura en el prompt; recuerde que la seguridad en entornos de producción depende de un rol de base de datos de solo lectura, no únicamente del prompt.
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,
)
Un pequeño ayudante imprime junto la pregunta, el Cypher generado, las filas de ejemplo y la respuesta final:
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
Pregunta sobre el roster de la cocina:
_ = 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).
Desglose más detallado del incidente:
_ = 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.
Próximos pasos a probar
Ejecutar la detección de comunidades (por ejemplo, Leiden) para agrupar los incidentes relacionados y precomputar resúmenes para preguntas que abarcan todo el corpus. Emparejar el grafo con un índice vectorial del texto de los informes brutos para que las preguntas vagas encuentren un nodo de entrada antes de continuar la exploración. Ampliar las listas de permisos con cuidado a medida que aparecen nuevos tipos de entidades, y evaluar el Cypher generado en un conjunto de preguntas reservado antes de ampliar los permisos de la base de datos.
Por qué la demostración se mantiene pequeña
Cuatro documentos son suficientes para practicar la extracción restringida, las consultas de vecindad, los caminos de múltiples saltos y la conversión NL→Cypher, con una protección visible de solo lectura. La historia de la cocina sigue siendo legible mientras se enseñan los mismos principios de GraphRAG que aplicaría a informes de incidentes o grafos de políticas: almacenar las relaciones una vez, consultarlas de diversas maneras y evitar que el modelo genere consultas de escritura contra datos en producción.
Lecturas relacionadas
- Construir un asistente de chat RAG para Wikipedia con LlamaIndex y ReAct — Indexar páginas de Wikipedia bajo demanda, incrustar fragmentos con un pequeño codificador y permitir que un agente ReAct llame a una herramienta de motor de consultas para que las respuestas de Chainlit estén basadas en evidencias recuperadas.
- Componer pipelines de LangChain con LCEL: lineal, paralelo y ramificado — Aprenda a conectar prompts, modelos y analizadores en pipelines de LangChain lineales, de múltiples etapas, paralelos y condicionales mediante el operador pipe, RunnableParallel y RunnableBranch.