GraphRAG de bout en bout avec FalkorDB, des embeddings et un routage LangGraph

Les grands modèles de langage répondent avec fluidité à partir des données d’entraînement, mais ils butent encore lorsque la question exige une structure nouvelle : qui a réalisé tel film, quels acteurs ont joué ensemble, quels liens relient deux entités. Les systèmes GraphRAG comblent cette lacune en combinant une recherche vectorielle sur du texte non structuré avec des requêtes graphiques explicites sur les relations, permettant ainsi à un LLM de raisonner à partir des deux approches.

Cette démarche permet de créer un flux de travail fonctionnel sur des données cinématographiques : FalkorDB en tant que stockage graphique, des embeddings Hugging Face MiniLM pour le rappel sémantique, des modèles hébergés sur Groq pour le routage et les réponses, des outils LangChain pour la récupération des données, ainsi que LangGraph pour orchestrer le processus. Il ne s’agit pas d’une démonstration simplifiée, mais d’un modèle que vous pouvez adapter à tout domaine où les relations sont aussi importantes que les paragraphes.

Pourquoi les graphes s’associent aux vecteurs

La recherche vectorielle permet de trouver des segments similaires. Les requêtes graphiques identifient des chemins structurés. Si vous demandez les films réalisés par Christopher Nolan, un bord graphique offre une précision supérieure à celle d’une estimation basée sur le voisin le plus proche. En demandant « des films de vol surprenants comme Inception », les vecteurs se révèlent particulièrement utiles. Les assistants réels ont besoin des deux approches, ainsi que d’un routeur capable de décider quel chemin suivre — ou quand utiliser une méthode hybride.

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

Environnement et FalkorDB

Créez un environnement virtuel, installez Jupyter si vous souhaitez utiliser des cahiers d’expérimentation, et fixez les versions des logiciels. L’interface de LangChain peut changer ; le fixation des versions fait partie de la reproductibilité.

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

Exécutez FalkorDB localement avec Docker, ou utilisez une instance gérée si cela convient à votre équipe. L’utilisation de Docker local permet de réduire les coûts liés aux expérimentations :

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

Installez exactement l’ensemble de bibliothèques utilisé pour tester ce flux de travail, après avoir supprimé les anciennes installations 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"

Chargement du dataset de style IMDB

Téléchargez un fichier CSV de films au format Top-1000, puis chargez-le à l’aide des importations et de la configuration de connexion attendues par votre notebook :

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)

Construire le graphe de connaissances

Créez des nœuds représentant les films avec des propriétés telles que le titre, l’année et la classification. Créez également des nœuds pour les réalisateurs et les acteurs. Connectez-les par des relations typées afin que les requêtes de type Cypher puissent répondre aux questions concernant le casting et l’équipe sans avoir à compter sur le fait qu’un embedding se souvienne de cette connexion.

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 pour le rappel sémantique

Utilisez all-MiniLM-L6-v2 à travers le package d’intégration moderne de Hugging Face plutôt que des chemins d’importation obsolètes :

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

Stockez les vecteurs à côté du graphe ou en coordination avec celui-ci, de manière à ce que les nœuds hybrides puissent récupérer les titres des films via une recherche sémantique, puis les développer à l’aide des relations.

GraphState pour LangGraph

Définissez un TypedDict (ou équivalent) GraphState qui contient la question de l’utilisateur, la décision de routage, les sous-queries, les résultats de recherche vectorielle, les réponses du graphe et la réponse finale. LangGraph conserve cet état entre les nœuds afin que les demandes ultérieures puissent réutiliser le contexte au lieu de devoir recommencer à partir d’une demande vide.

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

Routage des requêtes

Toute question ne doit pas être traitée par tous les sous-systèmes. Un routeur détermine si la question est principalement relationnelle, principalement sémantique, hybride ou hors champ. Cela permet de réduire la latence et d’éviter des contextes contradictoires.

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"

Décomposeur

Lorsque le parcours hybride est activé, un décomposeur divise la question de l’utilisateur en sous-queries que les nœuds vectoriels et graphiques peuvent traiter :

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}

Nœud de recherche vectorielle

Mettez en œuvre une recherche de similarité à l’aide des outils de récupération actuels 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"],
    }

Nœud de QA graphique

Traduisez les sous-questions relationnelles en requêtes graphiques. Pour une demande de style filmographie d’un réalisateur, le graphe doit retourner des titres reliés par des arêtes « réalisé par » plutôt qu’une supposition reformulée :

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}

QA graphique avec contexte vectoriel

Le nœud hybride prend en charge les titres récupérés sémantiquement et les étend avec une structure graphique — distribution, équipe, films associés — avant que l’LLM n’écrive une réponse finale :

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

Nœud de réponse

Normalisez la réponse finale en langage naturel, ajoutez des informations sur l’origine lorsque c’est possible, et supprimez les champs intermédiaires que vous ne souhaitez pas divulguer au client :

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}

Organiser le flux de travail LangGraph

Connectez les nœuds routeur, décomposeur, vectoriel, graphique, hybride et de réponse avec des arêtes conditionnelles correspondant aux étiquettes du parcours :

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

Visualisez le graphe à des fins de documentation et d’intégration :

from IPython.display import Image, display

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

Test et console interactive

Exécutez des questions scriptées couvrant les cas relationnels, sémantiques et hybrides :

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

Puis utilisez une boucle de console simple pour une exploration manuelle :

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

Résultats à vérifier

Vérifiez que les requêtes de type Nolan renvoient des titres ancrés dans le graphe, que les requêtes de type vibe renvoient des films sémantiquement pertinents, et que les requêtes hybrides citent les deux types de preuves. Lorsque les réponses s’écartent, examinez les enregistrements d’état entre les nœuds plutôt que seulement la chaîne finale.

Résumé et conseils d’adaptation

GraphRAG constitue autant un problème de routage qu’un problème de stockage. FalkorDB (ou tout autre base de données graphiques à faible latence) vous fournit une structure structurée. Les embeddings permettent un rappel plus approximatif des informations. LangGraph rend le flux de contrôle suffisamment explicite pour être testé. Remplacez les films par des éléments propres à votre domaine — pièces et fournisseurs, patients et protocoles, services et dépendances — tout en conservant les mêmes rôles de nœuds : router, récupérer, étendre, répondre.

Fixez les versions, conservez un ensemble d’questions types par type de route, et enregistrez les transitions d’état. Cette discipline opérationnelle est plus importante que le jeu de données de démonstration à partir duquel vous avez commencé.

Notes de conception résistantes à l’environnement de production

La résolution d’entités dominera les déploiements réels. Les titres de films sont relativement simples à gérer ; en revanche, les noms d’entreprises et les codes produits ne le sont pas. Planifiez une étape de normalisation avant la création des nœuds, et conservez un mappage entre les chaînes de caractères brutes et les IDs canoniques. Sans cela, les requêtes sur le graphe deviennent fragmentées et les résultats vectoriels indiquent des doublons.

L’évolution du schéma nécessite des migrations, tout comme en SQL. Ajouter un nouveau type de relation est simple dans un notebook, mais difficile une fois que les tableaux de bord dépendent des anciennes relations. Versionnez votre schéma de graphe et stockez-le à côté du code du flux de travail.

Les budgets de latence doivent être définis explicitement. Les appels aux LLM via un routeur ajoutent du temps avant même que la récupération ne commence. Pour des applications à fort taux de requêtes, envisagez l’utilisation de règles ou de petits classificateurs pour les schémas relationnels évidents (« qui a réalisé X »), et réservez le routage par LLM aux textes ambigus.

L’évaluation doit noter l’exactitude du parcours séparément de la qualité de la réponse. Une réponse parfaite mais sur le mauvais parcours consomme quand même des ressources et échouera en cas de panne d’un sous-système. Testez le système en désactivant soit le graphe, soit l’index vectoriel, afin de vérifier que le flux de travail se dégrade réellement.

La sécurité est importante : la génération des requêtes sur le graphe doit être contrôlée. Préférez les modèles paramétrés aux requêtes Cypher libres formulées à partir d’un modèle. La même règle appliquée aux outils SQL s’applique ici également.

Contrôle des coûts : mettez en cache les représentations vectorielles pour les documents stables, mettez en cache les décisions de routage pour des questions identiques pendant une durée de vie courte, et évitez l’expansion hybride lorsque le graphe pur fournit déjà des entités à haute fiabilité.

Lorsque vous expliquez cela lors d’un entretien, dessinez d’abord l’objet d’état, puis les arêtes. Les interviewers comprennent plus rapidement le flux de contrôle que les noms de packages. Mentionnez FalkorDB pour la vitesse des graphes de propriétés, MiniLM pour les embeddings locaux, et Groq uniquement comme un hôte LLM possible — votre architecture ne doit pas dépendre d’un seul fournisseur d’inférence.

Enfin, documentez les messages d’erreur que les utilisateurs verront lorsque le graphique est vide ou que le routeur refuse de fonctionner. Ce sont ces hallucinations silencieuses qui font que les démonstrations GraphRAG perdent en crédibilité dès la première semaine. Maintenez les éléments au niveau des routes en état valide avant d’élargir le schéma du graphique à de nouveaux types d’entités. Maintenez les éléments au niveau des routes en état valide avant d’élargir le schéma du graphique à de nouveaux types d’entités. Maintenez les éléments au niveau des routes en état valide avant d’élargir le schéma du graphique à de nouveaux types d’entités. Maintenez les éléments au niveau des routes en état valide avant d’élargir le schéma du graphique à de nouveaux types d’entités. Maintenez les éléments au niveau des routes en état valide avant d’élargir le schéma du graphique à de nouveaux types d’entités. Maintenez les éléments au niveau des routes en état valide avant d’élargir le schéma du graphique à de nouveaux types d’entités. Maintenez les éléments au niveau des routes en état valide avant d’élargir le schéma du graphique à de nouveaux types d’entités. Maintenez les éléments au niveau des routes en état valide avant d’élargir le schéma du graphique à de nouveaux types d’entités. Maintenez la route

Maintenez les éléments au niveau e en vert avant d’élargir le schéma du graphique à de nouveaux types d’entités.Maintenez les éléments de niveau route en vert avant d’élargir le schéma du graphique à de nouveaux types d’entités.Maintenez les éléments de configuration au niveau des routes en état vert avant d’élargir le schéma du graphique à de nouveaux types d’entités.

• Graphes des crimes financiers multijoueurs avec LangChain, LangGraph et LangSmith — Affectation des superviseurs, vérifications parallèles, sous-graphes, interruptions, évaluations et mécanismes de contrôle pour des workflows de niveau enquête.
• Recherche sémantique éphémère avec LangGraph lorsque l’on ne doit pas conserver des vecteurs — Des embeddings limités à la durée de la session qui disparaissent à leur fermeture établissent des limites claires — et permettent un passage ultérieur vers Redis ou Azure AI Search en modifiant simplement l’adaptateur de stockage.