Add FalkorDB Knowledge Graph with GraphRAG for LEGI, JADE, and BOFIP

.env.example

@@ -19,6 +19,11 @@ QDRANT_HOST=localhost
 QDRANT_PORT=6333
 QDRANT_API_KEY= # Optional: set for Qdrant Cloud or when API key auth is enabled
 
+FALKORDB_HOST=localhost
+FALKORDB_PORT=6379
+FALKORDB_USERNAME= # Optional: leave empty if authentication is disabled
+FALKORDB_PASSWORD= # Optional: leave empty if authentication is disabled
+
 AIRFLOW_UID= echo $(id -u)
 _AIRFLOW_WWW_USER_USERNAME=airflow
 _AIRFLOW_WWW_USER_PASSWORD=airflow

airflow_config/dags/knowledge_graph.py

@@ -0,0 +1,49 @@
+from datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+from notifier.notifications_template import (
+    get_failure_notifier,
+    get_start_notifier,
+    get_success_notifier,
+)
+
+from database import init_graph_schema, populate_graph_from_postgres
+
+default_args = {
+    "owner": "airflow",
+    "start_date": datetime(2025, 8, 1),
+    "retries": 1,
+    "retry_delay": timedelta(minutes=5),
+}
+
+with DAG(
+    "KNOWLEDGE_GRAPH",
+    default_args=default_args,
+    schedule=None,
+    catchup=False,
+    max_active_runs=1,
+    description=(
+        "FalkorDB knowledge graph initialisation and back-fill DAG. "
+        "Initialises the graph schema and populates it from data already "
+        "stored in PostgreSQL (LEGI, JADE, BOFIP)."
+    ),
+    tags=["mediatech", "knowledge_graph", "falkordb", "graphrag"],
+) as dag:
+    init_schema = PythonOperator(
+        task_id="init_graph_schema",
+        python_callable=init_graph_schema,
+        on_execute_callback=get_start_notifier(),
+        on_success_callback=get_success_notifier(),
+        on_failure_callback=get_failure_notifier(),
+    )
+
+    backfill_graph = PythonOperator(
+        task_id="populate_graph_from_postgres",
+        python_callable=populate_graph_from_postgres,
+        on_execute_callback=get_start_notifier(),
+        on_success_callback=get_success_notifier(),
+        on_failure_callback=get_failure_notifier(),
+    )
+
+    init_schema >> backfill_graph

config/__init__.py

@@ -7,6 +7,11 @@
     DATA_GOUV_DATASETS_CATALOG_DATA_FOLDER,
     DOLE_DATA_FOLDER,
     EMBEDDING_MODEL,
+    FALKORDB_GRAPH_NAME,
+    FALKORDB_HOST,
+    FALKORDB_PASSWORD,
+    FALKORDB_PORT,
+    FALKORDB_USERNAME,
     HF_TOKEN,
     LEGI_DATA_FOLDER,
     LLM_MODEL,

config/config.py

@@ -13,13 +13,17 @@
     postgres_port = "5432"
     qdrant_host = "qdrant"
     qdrant_port = "6333"
+    falkordb_host = "falkordb"
+    falkordb_port = "6379"
 else:
     # Locally, using relative paths
     base_path = "."
     postgres_host = os.getenv("POSTGRES_HOST", "localhost")
     postgres_port = os.getenv("POSTGRES_PORT", "5433")
     qdrant_host = os.getenv("QDRANT_HOST", "localhost")
     qdrant_port = os.getenv("QDRANT_PORT", "6333")
+    falkordb_host = os.getenv("FALKORDB_HOST", "localhost")
+    falkordb_port = os.getenv("FALKORDB_PORT", "6379")
 
 
 # PostgreSQL configuration
@@ -35,6 +39,13 @@
 QDRANT_URL = f"http://{qdrant_host}:{qdrant_port}"
 QDRANT_API_KEY = os.getenv("QDRANT_API_KEY", None)
 
+# FalkorDB configuration
+FALKORDB_HOST = falkordb_host
+FALKORDB_PORT = int(falkordb_port)
+FALKORDB_USERNAME = os.getenv("FALKORDB_USERNAME", None) or None
+FALKORDB_PASSWORD = os.getenv("FALKORDB_PASSWORD", None) or None
+FALKORDB_GRAPH_NAME = "frenchadmin"
+
 BASE_PATH = base_path
 
 # Paths for configurations and data history

database/__init__.py

@@ -11,3 +11,15 @@
     split_legi_table,
     sync_obsolete_doc_ids,
 )
+from .graph_manage import (
+    build_graphrag_knowledge_graph,
+    close_graph_connection,
+    init_graph_schema,
+    populate_graph_from_postgres,
+    upsert_bofip_chunk,
+    upsert_bofip_node,
+    upsert_jade_chunk,
+    upsert_jade_node,
+    upsert_legi_chunk,
+    upsert_legi_node,
+)

database/database_manage.py

@@ -445,6 +445,24 @@ def create_all_tables(model=EMBEDDING_MODEL, delete_existing: bool = False):
                         )
                     """)
 
+                elif table_name.lower() == "bofip":
+                    cursor.execute(f"""
+                        CREATE TABLE BOFIP (
+                            chunk_id TEXT PRIMARY KEY,
+                            doc_id TEXT NOT NULL,
+                            chunk_index INTEGER NOT NULL,
+                            chunk_xxh64 TEXT NOT NULL,
+                            nature TEXT,
+                            category TEXT,
+                            title TEXT,
+                            date TEXT,
+                            text TEXT,
+                            chunk_text TEXT,
+                            "embeddings_{model_name}" vector({embedding_size}),
+                            UNIQUE(chunk_id)
+                        )
+                    """)
+
                 # Create HNSW index for vector similarity search
                 try:
                     if table_name.lower() not in CONFIG_TABLES: