GraphRAG de extremo a extremo con FalkorDB, embeddings y enrutamiento LangGraph

Los modelos de lenguaje grandes responden con fluidez a partir de los datos de entrenamiento, pero todavía tropiezan cuando la pregunta requiere una estructura nueva: quién dirigió qué película, qué actores aparecen como coprotagonistas, qué enlaces conectan a dos entidades. Los sistemas GraphRAG resuelven esta deficiencia al combinar búsquedas vectoriales en texto no estructurado con consultas explícitas a grafos sobre relaciones, permitiendo así que un LLM razone con base en ambos.

Esta guía crea un flujo de trabajo funcional con datos de películas: FalkorDB como almacén de grafos, embeddings MiniLM de Hugging Face para la recuperación semántica, modelos alojados en Groq para el enrutamiento y las respuestas, herramientas de LangChain para la recuperación de información, y LangGraph para orquestar el proceso. El objetivo no es una demostración simplificada, sino un patrón que se puede adaptar a cualquier área donde las relaciones sean tan importantes como los párrafos.

Por qué los grafos se combinan con vectores

La búsqueda vectorial encuentra fragmentos similares. Las consultas de grafo localizan rutas específicas. Si pides películas dirigidas por Christopher Nolan, un enlace de grafo resulta más preciso que una estimación basada en el vecino más cercano. Si buscas “películas de robo alucinantes como Inception”, los vectores demuestran su eficacia. Los asistentes reales necesitan ambos métodos, además de un router que decida qué ruta seguir o cuándo utilizar una versión híbrida.

MATCH (d:Director {name: 'Christopher Nolan'})-[:DIRECTED]->(m:Movie) RETURN m.title
MATCH (m:Movie)-[r]->(d:Director) RETURN m, r, d

Entorno y FalkorDB

Crea un entorno virtual, instala Jupyter si deseas usar un cuaderno de notas exploratorio, y fija las versiones. La interfaz de LangChain puede cambiar; fijar las versiones es clave para la reproducibilidad.

$ pip install jupyterlab
$ jupyter lab

# 1. SETUP WSL (Run in Windows PowerShell as Administrator)
wsl --install
# Restart your computer if prompted, then create your Linux username/password.
# Inside the WSL/Ubuntu terminal, update packages:
sudo apt update && sudo apt upgrade -y

# 2. SET ENVIRONMENT VARIABLES (Paste your keys from Groq & Hugging Face)
export GROQ_API_KEY="your_groq_api_key_here"
export HUGGINGFACEHUB_API_TOKEN="your_hf_token_here"

# 3. ENVIRONMENT & DEPENDENCY SETUP
# Install Python pip and virtual environment tools
sudo apt install python3-pip python3-venv -y

# Create and activate a project directory
mkdir movie-graphrag && cd movie-graphrag
python3 -m venv venv
source venv/bin/activate


# Verify everything is correctly configured
echo "Groq Key: $GROQ_API_KEY"
echo "HF Token: $HUGGINGFACEHUB_API_TOKEN"

Ejecuta FalkorDB localmente con Docker, o utiliza una instancia gestionada si eso se adapta a tu equipo. Usar Docker local mantiene los experimentos económicos:

docker run -p 6379:6379 -p 3000:3000 -it --rm -v ./data:/data falkordb/falkordb:edge

Instala exactamente el conjunto de bibliotecas con las que se probó este flujo de trabajo, después de eliminar instalaciones anteriores de LangChain:

pip uninstall langchain langchain-core langchain-community langchain-groq \
              langchain-huggingface langgraph -y

pip install falkordb pandas pydantic sentence-transformers
pip install "langchain==0.3.26"
pip install "langchain-core==0.3.68"
pip install "langchain-community==0.3.23"
pip install "langchain-groq==0.2.5"
pip install "langchain-huggingface==0.1.2"
pip install "langgraph==0.3.34"

Cargar el conjunto de datos al estilo IMDB

Descargue un archivo CSV con películas al estilo Top-1000 y cárguelo utilizando las importaciones y la configuración de conexión que espera su cuaderno de notas:

import os, re, sys, logging
import pandas as pd
from typing import List, TypedDict, Literal
from falkordb import FalkorDB
from langchain_community.document_loaders.csv_loader import CSVLoader
from langchain_community.vectorstores.falkordb_vector import FalkorDBVector
from langchain_community.graphs import FalkorDBGraph
from langchain_community.chains.graph_qa.falkordb import FalkorDBQAChain
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.output_parsers import PydanticToolsParser
from pydantic import BaseModel, Field          # pydantic v2 — not langchain_core.pydantic_v1
from langchain_huggingface import HuggingFaceEmbeddings  # not langchain.embeddings (deprecated)
from langchain_groq import ChatGroq
from langgraph.graph import END, StateGraph

GROQ_API_KEY = "your_groq_api_key"
CSV_PATH     = "imdb_top_1000.csv"
os.environ["HF_TOKEN"] = "your_hf_token"

logging.basicConfig(stream=sys.stdout, level=logging.WARNING)

# Connect to FalkorDB and select the graph
db    = FalkorDB(host='localhost', port=6379)
graph = db.select_graph('imdb')

data = pd.read_csv(CSV_PATH)

Construir el grafo de conocimiento

Cree nodos para películas con propiedades como título, año y certificación. Crea también nodos para directores y actores. Conéctelos mediante relaciones tipadas de modo que las consultas al estilo Cypher puedan responder a preguntas sobre el reparto y el equipo sin depender de que un embedding recuerde esa conexión.

def clean_column(column_data):
    """Remove special characters that would break Cypher queries."""
    return column_data.apply(
        lambda x: re.sub(r"[^a-zA-Z0-9\s]", "", x) if pd.notnull(x) else 'NA'
    )

for col in ['Series_Title', 'Director', 'Certificate', 'Runtime', 'Genre',
            'Overview', 'Star1', 'Star2', 'Star3', 'Star4']:
    data[col] = clean_column(data[col])

data['IMDB_Rating']   = data['IMDB_Rating'].fillna('NA')
data['Released_Year'] = pd.to_numeric(data['Released_Year'], errors='coerce').fillna(-1)

star_columns = ['Star1', 'Star2', 'Star3', 'Star4']

for index, row in data.iterrows():
    movie_title   = row['Series_Title']
    director_name = row['Director']
    released_year = int(row['Released_Year']) if row['Released_Year'] != -1 else None

    if released_year is None:
        print(f"Skipping '{movie_title}' — invalid year.")
        continue

    movie_node = (
        f"MERGE (m:Movie {{"
        f"title: '{movie_title}', year: {released_year}, "
        f"certificate: '{row['Certificate']}', runtime: '{row['Runtime']}', "
        f"genre: '{row['Genre']}', imdb_rating: {row['IMDB_Rating']}, "
        f"description: '{row['Overview']}'"
        f"}})"
    )
    director_node     = f"MERGE (d:Director {{name: '{director_name}'}})"
    directed_relation = "MERGE (d)-[:DIRECTED]->(m)"

    star_nodes, starred_relations = "", ""
    for s, star_col in enumerate(star_columns, start=1):
        star_name = row[star_col]
        if star_name != 'NA':
            star_nodes        += f"MERGE (s{s}:Star {{name: '{star_name}'}})\n"
            starred_relations += f"MERGE (s{s})-[:STARRED_IN]->(m)\n"

    cypher_query = (
        f"{movie_node} {director_node} {directed_relation} "
        f"{star_nodes.strip()}\n{starred_relations.strip()}"
    )
    graph.query(cypher_query)

print("Knowledge Graph created.")

Embeddings para la recuperación semántica

Utilice all-MiniLM-L6-v2 a través del paquete de integración moderno de Hugging Face en lugar de rutas de importación obsoletas:

embedding_model = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

result = graph.query(
    "MATCH (m:Movie) RETURN m.title AS title, m.description AS description"
)

documents = [
    Document(
        page_content=row[1] if row[1] else "",
        metadata={"title": row[0], "description": row[1]}
    )
    for row in result.result_set if row[1]
]

vector_index = FalkorDBVector(host='localhost', port=6379, embedding=embedding_model)
vector_index.add_documents(documents)
print("Vector index created.")

Almacene los vectores junto con el grafo o en coordinación con él, de modo que los nodos híbridos puedan obtener títulos de películas a partir de búsquedas semánticas y luego ampliar esa información mediante relaciones.

GraphState para LangGraph

Se define un TypedDict (o equivalente) GraphState que contiene la pregunta del usuario, la decisión de ruta, las subconsultas, los resultados del vector, las respuestas del grafo y la respuesta final. LangGraph mantiene ese estado entre nodos para que las consultas posteriores puedan reutilizar el contexto en lugar de reiniciar desde una instrucción vacía.

class GraphState(TypedDict):
    """
    Tracks state across the entire LangGraph workflow.
    Every node reads from and writes to this dictionary.
    """
    question:            str
    documents:           object
    movie_titles:        List[str]
    subqueries:          object
    prompt_with_context: object

Ruteo de consultas

No toda pregunta debe llegar a todos los subsistemas. Un router clasifica si la pregunta es principalmente relacional, principalmente semántica, híbrida o está fuera del alcance. Esto reduce la latencia y disminuye los contextos contradictorios.

llm = ChatGroq(groq_api_key=GROQ_API_KEY, model_name="llama-3.3-70b-versatile")

class RouteQuery(BaseModel):
    """Route a user query to the most relevant datasource."""
    datasource: Literal["vector search", "graph query"] = Field(
        ...,
        description="Route to 'vector search' for similarity queries, "
                    "'graph query' for relationship/fact queries.",
    )

structured_llm_router = llm.with_structured_output(RouteQuery)

route_system_prompt = """You are an expert at routing user questions.

Route to VECTOR SEARCH for questions involving:
- Similarity (similar, related, like, closest, relevant)
- Themes or topics (movies about space, about love, etc.)
- "Find movies similar to X"

Route to GRAPH QUERY for questions involving:
- Specific facts (who directed X, what year was X released)
- Specific people (movies by Christopher Nolan)
- Filtering by property (movies from 1990, Action movies)
"""

route_prompt    = ChatPromptTemplate.from_messages([
    ("system", route_system_prompt),
    ("human",  "{question}"),
])
question_router = route_prompt | structured_llm_router

def route_question(state: GraphState):
    source = question_router.invoke({"question": state["question"]})
    if source.datasource == "vector search":
        return "decomposer"
    else:
        return "graph_qa"

Descompresor

Cuando se sigue el camino híbrido, un descompresor divide la pregunta del usuario en subconsultas que los nodos vectoriales y de grafo pueden procesar:

class SubQuery(BaseModel):
    """A single sub-query decomposed from the original question."""
    sub_query: str = Field(
        ..., description="A unique paraphrasing of the original question.",
    )

decompose_system = """You are an expert at breaking down complex questions.
Given a user question, split it into exactly two sub-queries:
  1. A query for vector similarity search (find related/similar items)
  2. A query for a graph database (find specific relationships or facts)

Example:
Question: Find movies directed by Christopher Nolan related to dreams.
sub_query 1: Find movies related to dreams or the subconscious.
sub_query 2: Find movies directed by Christopher Nolan.
"""

decompose_prompt = ChatPromptTemplate.from_messages([
    ("system", decompose_system),
    ("human",  "{question}"),
])

llm_with_tools = llm.bind_tools([SubQuery])
parser         = PydanticToolsParser(tools=[SubQuery])
query_analyzer = decompose_prompt | llm_with_tools | parser

def decomposer(state: GraphState):
    question   = state["question"]
    subqueries = query_analyzer.invoke({"question": question})
    return {"subqueries": subqueries, "question": question}

Nodo de búsqueda vectorial

Se implementa una búsqueda por similitud con las herramientas de recuperación actuales de LangChain:

def vector_search(state: GraphState):
    question  = state["question"]
    retriever = vector_index.as_retriever(search_kwargs={"k": 3})

    qa_prompt = ChatPromptTemplate.from_messages([
        ("system",
         "You are a movie expert. Use the context below to answer the question.\n\n"
         "Context:\n{context}"),
        ("human", "{input}"),
    ])

    combine_docs_chain = create_stuff_documents_chain(llm, qa_prompt)
    retrieval_chain    = create_retrieval_chain(retriever, combine_docs_chain)

    result      = retrieval_chain.invoke({"input": question})
    source_docs = result.get("context", [])

    extracted_data, movie_titles = [], []
    for doc in source_docs:
        title       = doc.metadata.get("title", "Unknown")
        description = doc.metadata.get("description", "")
        extracted_data.append({"title": title, "description": description})
        movie_titles.append(title)

    return {
        "movie_titles": movie_titles,
        "documents":    extracted_data,
        "question":     question,
        "subqueries":   state["subqueries"],
    }

Nodo de preguntas y respuestas en grafo

Traduzca las subpreguntas relacionales a consultas de grafo. En el caso de una solicitud al estilo de la filmografía de un director, el grafo debe devolver títulos conectados mediante aristas de “dirigido por”, en lugar de una suposición reescrita:

def graph_qa(state: GraphState):
    question      = state["question"]
    graph_wrapper = FalkorDBGraph(database="imdb", host="localhost", port=6379)

    chain  = FalkorDBQAChain.from_llm(
        llm,
        graph=graph_wrapper,
        verbose=True,
        allow_dangerous_requests=True,
    )
    result = chain.invoke({"query": question})
    answer = result.get("result", str(result))

    return {"documents": answer, "question": question}

Preguntas y respuestas con grafo y contexto vectorial

El nodo híbrido toma los títulos recuperados semánticamente y los expande con estructura de grafo: reparto, equipo técnico, películas relacionadas—antes de que el LLM escriba la respuesta final:

def graph_qa_with_context(state: GraphState):
    titles      = state["movie_titles"]
    question    = state["question"]
    input_query = (
        f"Return the title, genre, director, and IMDB rating of movies "
        f"with these titles: {str(titles)}"
    )

    graph_wrapper = FalkorDBGraph(database="imdb", host="localhost", port=6379)
    chain  = FalkorDBQAChain.from_llm(
        llm,
        graph=graph_wrapper,
        verbose=True,
        allow_dangerous_requests=True,
        return_intermediate_steps=True,
    )
    result  = chain.invoke({"query": input_query})
    context = ""
    steps   = result.get("intermediate_steps", [])
    if len(steps) > 1:
        context = str(steps[1].get("context", ""))

    return {
        "documents": context,
        "subqueries": state["subqueries"],
        "question":   question,
    }

Nodo de respuesta

Normalice la respuesta final en lenguaje natural, adjunte la procedencia cuando sea posible y elimine los campos intermedios que no desea revelar al cliente:

def respond(state: GraphState):
    question = state["question"]
    context  = state["documents"]

    respond_prompt = ChatPromptTemplate.from_messages([
        ("system",
         "You are a helpful movie expert assistant.\n"
         "Use the context below to answer the user's question clearly and concisely.\n\n"
         "Context:\n{context}"),
        ("human", "{question}"),
    ])

    answer = (respond_prompt | llm).invoke({"context": context, "question": question})
    return {"documents": answer.content}

Conectar los nodos del flujo de trabajo LangGraph

Conecte los nodos de enrutador, descompresor, vector, grafo, híbrido y respuesta con aristas condicionales que coincidan con las etiquetas de la ruta:

workflow = StateGraph(GraphState)

# Register all nodes
workflow.add_node("graph_qa",              graph_qa)
workflow.add_node("decomposer",            decomposer)
workflow.add_node("vector_search",         vector_search)
workflow.add_node("graph_qa_with_context", graph_qa_with_context)
workflow.add_node("respond",               respond)

# Conditional entry point — route_question decides the first node
workflow.set_conditional_entry_point(
    route_question,
    {
        "decomposer": "decomposer",
        "graph_qa":   "graph_qa",
    },
)

# Hybrid path
workflow.add_edge("decomposer",            "vector_search")
workflow.add_edge("vector_search",         "graph_qa_with_context")
workflow.add_edge("graph_qa_with_context", "respond")
workflow.add_edge("respond",               END)

# Direct graph path
workflow.add_edge("graph_qa", END)

app = workflow.compile()

Visualice el grafo para documentación e inducción al uso:

from IPython.display import Image, display

try:
    display(Image(app.get_graph().draw_mermaid_png()))
except Exception:
    pass

Prueba y consola interactiva

Ejecuta preguntas basadas en scripts que abarcan casos relacionales, semánticos e híbridos:

# Graph query — specific fact question
result1 = app.invoke({"question": "Which movies did Christopher Nolan direct?"})
print(result1["documents"])

Luego utiliza un bucle simple en la consola para una exploración manual:

print("\n" + "=" * 60)
print("INTERACTIVE MODE — type a question (or 'quit' to exit)")
print("=" * 60)

while True:
    user_question = input("\nYour question: ").strip()
    if user_question.lower() in ("quit", "exit", "q"):
        print("Goodbye!")
        break
    if not user_question:
        continue
    try:
        result = app.invoke({"question": user_question})
        print("\nANSWER:", result["documents"])
    except Exception as e:
        print(f"Error: {e}")

Resultados que vale la pena verificar

Verifica que las consultas de tipo director al estilo Nolan devuelvan títulos basados en el grafo, que las consultas de tipo vibe devuelvan películas semánticamente coherentes, y que las consultas híbridas mencionen ambos tipos de evidencia. Cuando las respuestas difieran, examina los registros de estado entre nodos en lugar de solo la cadena final.

Resumen y consejos de adaptación

GraphRAG es tanto un problema de enrutamiento como uno de almacenamiento. FalkorDB (o cualquier base de datos gráfica de baja latencia) proporciona una estructura definida. Las embeddings permiten obtener un recuerdo más aproximado. LangGraph hace que el flujo de control sea lo suficientemente explícito como para poder probarlo. Sustituya las películas por los elementos de su dominio: piezas y proveedores, pacientes y protocolos, servicios y dependencias, manteniendo los mismos roles de nodos: enrutamiento, recuperación, expansión y respuesta.

Fije las versiones, mantenga un conjunto de preguntas clave para cada tipo de ruta y registre las transiciones de estado. Esa disciplina operativa es más importante que el conjunto de datos de demostración desde el cual se partió.

Notas de diseño que resisten la integración en entornos reales

La resolución de entidades dominará las implementaciones reales. Los títulos de películas son relativamente simples; los nombres de empresas y los códigos SKU de productos no lo son. Planifique un paso de normalización antes de crear nodos y mantenga una relación entre cadenas de texto sin procesar e IDs canónicos. Sin esto, las consultas al grafo se fragmentan y los resultados vectoriales apuntan a duplicados.

La evolución del esquema requiere migraciones al igual que SQL. Añadir un nuevo tipo de relación es sencillo en un cuaderno de notas, pero difícil una vez que los paneles dependen de conexiones antiguas. Versione su esquema de grafo y guárdelo junto al código del flujo de trabajo.

Los presupuestos de latencia deben ser explícitos. Las llamadas a modelos LLM de enrutamiento añaden tiempo incluso antes de que comience la recuperación de datos. Para aplicaciones con altas tasas de consultas, considere reglas o pequeños clasificadores para patrones relacionales obvios (“¿quién dirigió X?”) y reserve el enrutamiento con modelos LLM para textos ambiguos.

La evaluación debe calificar la precisión de la ruta por separado de la calidad de la respuesta. Una respuesta perfecta en la ruta incorrecta sigue consumiendo recursos y fallará cuando uno de los subsistemas esté inoperativo. Realice pruebas de caos desactivando el grafo o el índice vectorial y confirme que el flujo de trabajo se degrade de manera realista.

La seguridad es importante: la generación de consultas al grafo debe estar restringida. Prefiera plantillas parametrizadas en lugar de Cypher sin formato definido generado a partir de un modelo. La misma regla que se aplica a las herramientas SQL también rige aquí.

Control de costos: almacene en caché las representaciones de documentos estables, guarde en caché las decisiones del router para preguntas idénticas durante un tiempo de vida corto, y evite la expansión híbrida cuando el grafo puro ya devuelva entidades de alta confiabilidad.

Al explicar esto en una entrevista, dibuje primero el objeto de estado y luego las aristas. Los entrevistadores siguen el flujo de control más rápido que los nombres de los paquetes. Mencione FalkorDB por su velocidad en grafos de propiedades, MiniLM para embeddings locales, y Groq solo como un posible host de LLM; su arquitectura no debe estar atada a un único proveedor de inferencia.

Finalmente, documente los mensajes de error que verán los usuarios cuando el gráfico esté vacío o el enrutador se niegue a funcionar. Las “alucinaciones silenciosas” son la causa por la cual las demostraciones de GraphRAG pierden confianza ya después de la primera semana. Mantenga los elementos a nivel de ruta en estado verde antes de expandir el esquema del gráfico con nuevos tipos de entidad. Mantenga los elementos a nivel de ruta en estado verde antes de expandir el esquema del gráfico con nuevos tipos de entidad. Mantenga los elementos a nivel de ruta en estado verde antes de expandir el esquema del gráfico con nuevos tipos de entidad. Mantenga los elementos a nivel de ruta en estado verde antes de expandir el esquema del gráfico con nuevos tipos de entidad. Mantenga los elementos a nivel de ruta en estado verde antes de expandir el esquema del gráfico con nuevos tipos de entidad. Mantenga los elementos a nivel de ruta en estado verde antes de expandir el esquema del gráfico con nuevos tipos de entidad. Mantenga los elementos a nivel de ruta en estado verde antes de expandir el esquema del gráfico con nuevos tipos de entidad. Mantenga los elementos a nivel de ruta en estado verde antes de expandir el esquema del gráfico con nuevos tipos de entidad. Mantenga la ruta

Mantener los elementos de nivel e en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos de nivel ruta en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos de nivel ruta en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos de nivel ruta en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos de nivel ruta en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos de nivel ruta en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos de nivel ruta en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos de nivel ruta en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos de nivel ruta en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos de nivel ruta en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos de nivel ruta en verde antes de expandir el esquema del gráfico a nuevos tipos de entidad. Mantener los elementos fi de nivel ruta en verde.Mantenga los elementos de nivel de ruta en color verde antes de expandir el esquema del gráfico a nuevos tipos de entidad.Mantenga los elementos a nivel de ruta en color verde antes de expandir el esquema del grafo en nuevos tipos de entidad.

• Grafos de Delitos Financieros Multiaagentes con LangChain, LangGraph y LangSmith — Enrutamiento por supervisor, verificaciones paralelas, subgrafos, interrupciones, evaluaciones y medidas de seguridad para flujos de trabajo de casos a nivel de investigación.
• Búsqueda semántica efímera con LangGraph cuando no se deben conservar vectores — Incrustaciones de ámbito de sesión que desaparecen al finalizar la sesión establecen límites claros, y permiten el posterior cambio a Redis o Azure AI Search como mera modificación del adaptador de almacenamiento.