Lesson 16.20: GraphRAG & Advanced Knowledge Retrieval

01 The Dilemma: Can Vector Search Answer Every Inquiry?

In previous lessons, we built a highly sophisticated RAG pipeline: text extraction, chunking, embeddings, vector databases, hybrid retrieval, reranking, and corrective evaluation. This pipeline excels at answering direct factual inquiries:

Now examine a realistic, multi-hop enterprise supply chain inquiry:

"Who is the official supplier of keyboards belonging to the product category managed by the Procurement Department, and what warranty covers them?"

To answer this question, no single document chunk suffices. The necessary facts are fragmented across disparate entities and corporate records:

Vector search operates on Semantic Similarity across isolated text blocks. It has no structural mechanism to navigate from a department to a category, from a category to a product SKU, and from a SKU to an authorized supplier. When facts reside in separated chunks, vector search cannot assemble this relational chain.

02 What is a Knowledge Graph? Nodes, Relationships, and Properties

A Knowledge Graph (KG) is a graph-structured data model that represents real-world entities and their conceptual relationships as an interconnected network of nodes and edges.

A labeled property graph consists of three fundamental primitives:

For instance, in our Byteion Assistant knowledge model, a factual relationship triplet (Subject-Predicate-Object) is structured as:

03 Knowledge Graph vs. Relational Database (SQL)

Coming from relational ecosystems like PostgreSQL, MySQL, and Laravel, engineers must understand the architectural distinction between relational tables and graph models:

Enterprise Coexistence: In modern architectures, relational databases are never abandoned. User authentication and financial ledgers reside in PostgreSQL, embeddings and chunks live in Qdrant, and organizational relational networks operate in Neo4j.

04 What is GraphRAG and How Does it Work?

GraphRAG is an advanced retrieval-augmented generation paradigm that leverages a structured Knowledge Graph alongside semantic vector search to augment language models with relational context.

05 When Should You Implement GraphRAG?

Introducing GraphRAG prematurely into a simple RAG application is an expensive anti-pattern. Evaluate the query complexity before opting for graph modeling:

06 How GraphRAG Works: Construction & Indexing Pipeline

A complete GraphRAG architecture decouples into two distinct lifecycles: 1. Indexing-Time Graph Construction and 2. Query-Time Retrieval.

07 Extracting Entities & Relationships with Pydantic

To transform unstructured text into validated knowledge triplets, we use Pydantic v2 schemas and OpenAI Structured Outputs:

# app/entity_extractor.py
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
from openai import OpenAI
import json

class ExtractedEntity(BaseModel):
    model_config = ConfigDict(extra="forbid")
    name: str = Field(description="Normalized canonical entity name")
    label: Literal["Company", "Department", "Category", "Product", "Supplier", "Policy"]
    description: str

class ExtractedRelationship(BaseModel):
    model_config = ConfigDict(extra="forbid")
    source_entity: str
    target_entity: str
    relation_type: Literal["HAS_DEPARTMENT", "MANAGES", "CONTAINS", "SUPPLIES", "COVERED_BY"]
    evidence_quote: str = Field(description="Exact sentence fragment supporting this relation")

class GraphExtractionResult(BaseModel):
    model_config = ConfigDict(extra="forbid")
    entities: list[ExtractedEntity]
    relationships: list[ExtractedRelationship]

def extract_knowledge_graph_triplets(client: OpenAI, model: str, chunk_text: str) -> GraphExtractionResult:
    system_prompt = (
        "You are an enterprise ontology extraction specialist. Extract strictly factual entities "
        "and relationships matching our predefined schema. Do NOT invent claims not explicitly in the text. "
        "Every relationship must include an exact evidence_quote from the input."
    )
    
    response = client.responses.parse(
        model=model,
        input=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"DOCUMENT CHUNK:\n{chunk_text}"}
        ],
        text_format=GraphExtractionResult
    )
    return response.output_parsed

08 Enterprise Knowledge Graph Schema Modeling

Allowing an LLM to invent arbitrary node labels or edge types produces an unmaintainable graph. One extraction might yield MANAGES, another CONTROLS, and another OPERATES. A strict Domain Schema is non-negotiable:

09 Neo4j Database Setup & Python Driver Configuration

The industry standard for property graph databases is Neo4j, powered by the declarative query language Cypher. Install the official driver:

pip install neo4j python-dotenv

# app/graph_db.py
import os
from neo4j import GraphDatabase
from dotenv import load_dotenv

load_dotenv()

URI = os.getenv("NEO4J_URI", "bolt://localhost:7687")
USER = os.getenv("NEO4J_USERNAME", "neo4j")
PASSWORD = os.getenv("NEO4J_PASSWORD")

if not PASSWORD:
    raise RuntimeError("NEO4J_PASSWORD environment variable is missing!")

# Connection pool driver instance
graph_driver = GraphDatabase.driver(URI, auth=(USER, PASSWORD), max_connection_lifetime=3600)
graph_driver.verify_connectivity()
print("✓ Neo4j Connection Established Successfully.")

10 Populating Knowledge Graphs with Cypher MERGE

Executing CREATE in Cypher produces duplicate nodes on every ingestion run. Instead, production architectures use MERGE, which locates existing records or initializes new entities idempotently:

# app/graph_writer.py
def upsert_enterprise_graph(driver, data: dict):
    """Upserts entities and relationships under tenant isolation."""
    query = """
    MERGE (company:Company {tenant_id: $tenant_id, company_id: $company_id})
    SET company.name = $company_name

    MERGE (dept:Department {tenant_id: $tenant_id, department_id: $dept_id})
    SET dept.name = $dept_name

    MERGE (cat:Category {tenant_id: $tenant_id, category_id: $cat_id})
    SET cat.name = $cat_name

    MERGE (prod:Product {tenant_id: $tenant_id, sku: $sku})
    SET prod.name = $prod_name

    MERGE (supp:Supplier {tenant_id: $tenant_id, supplier_id: $supp_id})
    SET supp.name = $supp_name

    MERGE (company)-[:HAS_DEPARTMENT]->(dept)
    MERGE (dept)-[:MANAGES]->(cat)
    MERGE (cat)-[:CONTAINS]->(prod)
    MERGE (supp)-[:SUPPLIES]->(prod)

    RETURN prod.sku AS sku
    """
    with driver.session() as session:
        result = session.run(query, data)
        return result.single()["sku"]

11 & 12 Multi-Hop Retrieval: Answering Complex Queries with Graph Traversals

We can now resolve our core compound question: "Which suppliers provide products within categories managed by the Procurement department?"

# app/graph_retriever.py
def multi_hop_procurement_query(driver, tenant_id: str, dept_name: str) -> list[dict]:
    query = """
    MATCH (dept:Department {tenant_id: $tenant_id, name: $dept_name})
          -[:MANAGES]->(cat:Category {tenant_id: $tenant_id})
          -[:CONTAINS]->(prod:Product {tenant_id: $tenant_id})
    MATCH (supp:Supplier {tenant_id: $tenant_id})-[:SUPPLIES]->(prod)
    RETURN 
        dept.name AS department,
        cat.name AS category,
        prod.name AS product,
        prod.sku AS sku,
        supp.name AS supplier
    """
    with driver.session() as session:
        records = session.run(query, {"tenant_id": tenant_id, "dept_name": dept_name})
        return [record.data() for record in records]

A single Cypher traversal navigates 3 distinct relationship hops, connecting previously fragmented facts that vector search could never reliably link.

13 Source Provenance: Grounding Graph Facts to Document Chunks

A fundamental vulnerability of knowledge graphs is that relationships can become detached from their original citations. When an auditor asks: "What verified document proves that Example Supplies manufactures this keyboard?", the system must produce the exact source citation.

We resolve this by embedding Provenance Tracking directly into the graph:

Every relationship stores source_chunk_id and extraction confidence properties, empowering language models to produce auditable page citations.

14 Entity Resolution: Eliminating Duplicate Nodes & Tenant Leaks

Corporate documents reference entities using diverse names: "ABC Corp", "ABC Corporation Ltd", and "ABC". Treating these as three separate nodes fragments the knowledge graph.

15 The Neo4j GraphRAG Python Package & SimpleKGPipeline

The official neo4j-graphrag package streamlines building knowledge graphs from text and executing hybrid graph retrievals:

# app/neo4j_pipeline.py
import asyncio
from neo4j_graphrag.llm import OpenAILLM
from neo4j_graphrag.embeddings import OpenAIEmbeddings
from neo4j_graphrag.experimental.pipeline.kg_builder import SimpleKGPipeline
from app.graph_db import graph_driver

llm = OpenAILLM(model_name="gpt-4o-mini")
embedder = OpenAIEmbeddings(model="text-embedding-3-small")

async def ingest_document_to_graph(text: str):
    pipeline = SimpleKGPipeline(
        llm=llm,
        driver=graph_driver,
        embedder=embedder,
        from_file=False
    )
    result = await pipeline.run_async(text=text)
    return result

Note: SimpleKGPipeline is currently marked experimental. Production deployments must place strict schema validation and tenant resolution guards in front of the builder.

16 Unifying Vector Search and Graph Traversals

The most robust architecture in production is Hybrid Graph Retrieval. Its execution flow operates as follows:

17 Microsoft GraphRAG: Local Search vs. Global Search

In their landmark framework, Microsoft introduced two complementary search strategies for GraphRAG:

18 Operational Constraints & Maintenance Costs of GraphRAG

While immensely capable, GraphRAG introduces significant operational overhead:

• High LLM Ingestion Costs: Extracting entity triplets across thousands of chunks requires substantial LLM compute.
• Graph Mutation Overhead: Updating or deleting documents requires identifying and pruning obsolete edges.
• The Open-World Assumption: The absence of an edge does not prove the relationship does not exist in reality.

19 Practical Project: FastAPI & Neo4j GraphRAG Service

The complete production FastAPI router and automated unit test suite for the GraphRAG pipeline are presented below:

# app/routers/graphrag_api.py
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
from app.auth import get_current_user, UserContext
from app.graph_db import graph_driver
from openai import OpenAI
import os

router = APIRouter(prefix="/v1/graphrag", tags=["GraphRAG"])
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

class MultiHopQueryRequest(BaseModel):
    department: str = Field(min_length=2, max_length=100)

@router.post("/query-suppliers")
async def query_procurement_suppliers(
    request: MultiHopQueryRequest,
    user: UserContext = Depends(get_current_user)
):
    query = """
    MATCH (dept:Department {tenant_id: $tenant_id, name: $dept_name})
          -[:MANAGES]->(cat:Category {tenant_id: $tenant_id})
          -[:CONTAINS]->(prod:Product {tenant_id: $tenant_id})
    MATCH (supp:Supplier {tenant_id: $tenant_id})-[:SUPPLIES]->(prod)
    RETURN 
        dept.name AS department,
        cat.name AS category,
        prod.name AS product,
        prod.sku AS sku,
        supp.name AS supplier
    """
    try:
        with graph_driver.session() as session:
            records = session.run(query, {
                "tenant_id": user.tenant_id,
                "dept_name": request.department
            })
            results = [r.data() for r in records]
            
        if not results:
            return {"status": "empty", "message": "No related products or suppliers found.", "data": []}
            
        return {"status": "success", "tenant_id": user.tenant_id, "data": results}
    except Exception as exc:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Graph retrieval failed: {str(exc)}"
        )

# tests/test_graphrag.py
import pytest
from unittest.mock import MagicMock

def test_tenant_isolation_in_graph_query():
    mock_session = MagicMock()
    mock_driver = MagicMock()
    mock_driver.session.return_value.__enter__.return_value = mock_session
    
    tenant_id = "tenant_test_123"
    dept_name = "Procurement"
    
    # Simulate multi-hop call
    query = "MATCH (d:Department {tenant_id: $tenant_id})..."
    mock_session.run(query, {"tenant_id": tenant_id, "dept_name": dept_name})
    
    mock_session.run.assert_called_with(query, {"tenant_id": "tenant_test_123", "dept_name": "Procurement"})

20 Knowledge Check: Scenario-Based Assessment

Test your technical understanding with the following 8 scenario-based questions: