Graph RAG: Enriching Retrieval with Knowledge Graphs

Standard RAG systems retrieve text chunks that are semantically similar to a query. This works well when the answer is contained in a single passage. It breaks down for questions that require connecting information across multiple documents, following relationships between entities, or reasoning over structured domain knowledge.

Graph RAG addresses this by combining a knowledge graph with vector retrieval. Instead of treating the knowledge base as a flat collection of text chunks, it represents entities and their relationships explicitly — and uses that graph structure to guide and enrich retrieval.

Where Standard RAG Fails

Consider a medical knowledge base containing documents about drugs, diseases, and interactions. A user asks:

“Which medications used to treat Type 2 diabetes have known interactions with ACE inhibitors?”

A vector retrieval system will retrieve passages mentioning diabetes and ACE inhibitors. But it cannot systematically traverse the relationship: Drug → treats → Disease AND Drug → interacts_with → Drug. Answering this question correctly requires structured relational reasoning that flat chunk retrieval cannot provide.

Other failure modes of standard RAG that Graph RAG addresses:

• Multi-hop reasoning: “Who is the manager of the department responsible for the project that customer X is associated with?”
• Entity disambiguation: “Apple” could be the fruit, the company, or a restaurant — a graph encodes which is which.
• Aggregation questions: “How many contracts were signed with vendors in Germany in 2023?” — requires traversal and counting.
• Relationship questions: “What is the relationship between Entity A and Entity B?”

What Is a Knowledge Graph

A knowledge graph is a directed graph of entities and typed relationships:

(GPT-4) -[developed_by]-> (OpenAI)
(OpenAI) -[founded_by]-> (Sam Altman)
(GPT-4) -[architecture]-> (Transformer)
(Transformer) -[introduced_in]-> (Attention Is All You Need)
(GPT-4) -[successor_of]-> (GPT-3.5)

Nodes are entities with properties. Edges are typed relationships. The graph can be queried structurally (Cypher, SPARQL) rather than only by semantic similarity.

Graph RAG Architecture

flowchart LR
  A[User Query] --> B[Query Analyzer]
  B --> C{Query Type}
  C -->|Semantic| D[Vector Retrieval]
  C -->|Relational| E[Graph Traversal\nCypher Query]
  C -->|Hybrid| F[Both Paths]
  D --> G[Fusion Layer]
  E --> G
  F --> G
  G --> H[Context Builder]
  H --> I[LLM]
  I --> J[Grounded Response]

The key addition over standard RAG is the graph traversal path. Depending on query analysis, the system routes to vector retrieval, graph traversal, or both.

Building the Knowledge Graph

Entity and relationship extraction

The first step is extracting a graph from source documents using an LLM:

from openai import OpenAI
from pydantic import BaseModel
from typing import List

client = OpenAI()

class Relationship(BaseModel):
    source: str
    source_type: str
    relation: str
    target: str
    target_type: str
    confidence: float

class ExtractionResult(BaseModel):
    entities: List[str]
    relationships: List[Relationship]

EXTRACTION_PROMPT = """
Extract entities and relationships from the following text.
For each relationship, provide:
- source entity and type
- relationship type (use active verb form: "treats", "inhibits", "founded_by")
- target entity and type
- confidence (0.0 to 1.0)

Text:
{text}

Return JSON following the schema provided.
"""

def extract_graph(text: str) -> ExtractionResult:
    response = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=[{"role": "user", "content": EXTRACTION_PROMPT.format(text=text)}],
        response_format=ExtractionResult,
    )
    return response.choices[0].message.parsed

Loading into Neo4j

from neo4j import GraphDatabase

class GraphStore:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def upsert_entity(self, name: str, entity_type: str, properties: dict = None):
        with self.driver.session() as session:
            session.run(
                f"""
                MERGE (e:{entity_type} )
                SET e += $props
                """,
                name=name,
                props=properties or {}
            )

    def upsert_relationship(self, source: str, relation: str, target: str,
                            source_type: str, target_type: str, confidence: float):
        with self.driver.session() as session:
            session.run(
                f"""
                MERGE (s:{source_type} )
                MERGE (t:{target_type} )
                MERGE (s)-[r:{relation.upper().replace(' ', '_')}]->(t)
                SET r.confidence = $confidence
                """,
                source=source, target=target, confidence=confidence
            )

    def ingest_extraction(self, result: ExtractionResult):
        for rel in result.relationships:
            self.upsert_entity(rel.source, rel.source_type)
            self.upsert_entity(rel.target, rel.target_type)
            self.upsert_relationship(
                rel.source, rel.relation, rel.target,
                rel.source_type, rel.target_type, rel.confidence
            )

Querying the Graph

LLM-generated Cypher queries

Convert natural language questions to Cypher queries using an LLM:

CYPHER_GENERATION_PROMPT = """
You are a Neo4j expert. Convert the user question to a Cypher query.

Graph schema:
- Nodes: Drug(name, mechanism), Disease(name, category), Protein(name, function)
- Relationships: TREATS(confidence), INHIBITS(dosage), INTERACTS_WITH(severity), TARGETS

Question: {question}

Return only the Cypher query, no explanation.
"""

def question_to_cypher(question: str, llm) -> str:
    return llm.generate(
        CYPHER_GENERATION_PROMPT.format(question=question),
        temperature=0
    ).strip()

def graph_retrieve(question: str, graph: GraphStore, llm) -> list[dict]:
    cypher = question_to_cypher(question, llm)
    try:
        with graph.driver.session() as session:
            result = session.run(cypher)
            return [dict(record) for record in result]
    except Exception as e:
        # Fallback: ask LLM to fix the query
        fixed_cypher = llm.generate(
            f"This Cypher query failed:\n{cypher}\n\nError: {e}\n\nFix it:"
        )
        with graph.driver.session() as session:
            result = session.run(fixed_cypher.strip())
            return [dict(record) for record in result]

Example Cypher queries

-- Multi-hop: drugs that treat diabetes AND interact with ACE inhibitors
MATCH (d:Drug)-[:TREATS]->(dis:Disease {name: "Type 2 Diabetes"})
MATCH (d)-[:INTERACTS_WITH]->(d2:Drug)-[:INHIBITS]->(p:Protein {name: "ACE"})
RETURN d.name as drug, d2.name as ace_inhibitor, dis.name as disease
ORDER BY d.name

-- Shortest path between two entities
MATCH path = shortestPath(
  (a:Entity {name: "OpenAI"})-[*..6]-(b:Entity {name: "GPT-4"})
)
RETURN path

-- Entity neighborhood for context building
MATCH (e:Drug {name: "Metformin"})-[r]-(neighbor)
RETURN type(r) as relation, neighbor.name as entity, labels(neighbor) as types
LIMIT 20

Hybrid Graph + Vector Retrieval

The most powerful approach combines both retrieval paths:

from sentence_transformers import SentenceTransformer
import chromadb

class GraphRAGRetriever:
    def __init__(self, graph: GraphStore, vector_collection, llm, embed_model):
        self.graph = graph
        self.vector_collection = vector_collection
        self.llm = llm
        self.embedder = SentenceTransformer(embed_model)

    def retrieve(self, question: str, top_k: int = 5) -> dict:
        # 1. Classify query
        query_type = self._classify_query(question)
        
        results = {"vector": [], "graph": []}
        
        # 2. Vector retrieval (always run)
        query_embedding = self.embedder.encode(question).tolist()
        vector_results = self.vector_collection.query(
            query_embeddings=[query_embedding], n_results=top_k
        )
        results["vector"] = vector_results["documents"][0]
        
        # 3. Graph retrieval (for relational queries)
        if query_type in ("relational", "multi-hop", "hybrid"):
            graph_results = graph_retrieve(question, self.graph, self.llm)
            results["graph"] = graph_results
        
        return results

    def _classify_query(self, question: str) -> str:
        prompt = f"""
        Classify this question as one of: semantic, relational, multi-hop, hybrid.
        - semantic: factual, can be answered from a single passage
        - relational: requires traversing relationships between entities
        - multi-hop: requires connecting 3+ entities across hops
        - hybrid: needs both document text and relationship structure
        
        Question: {question}
        Answer with one word:
        """
        return self.llm.generate(prompt, temperature=0).strip().lower()

Context Assembly

Once you have both vector chunks and graph results, assemble them into a coherent context for the LLM:

def build_context(vector_chunks: list[str], graph_results: list[dict]) -> str:
    context_parts = []
    
    if vector_chunks:
        context_parts.append("## Relevant Document Passages\n")
        for i, chunk in enumerate(vector_chunks, 1):
            context_parts.append(f"[DOC-{i}] {chunk}\n")
    
    if graph_results:
        context_parts.append("\n## Knowledge Graph Facts\n")
        for result in graph_results:
            facts = ", ".join(f"{k}: {v}" for k, v in result.items())
            context_parts.append(f"- {facts}")
    
    return "\n".join(context_parts)

SYSTEM_PROMPT = """
You are an assistant with access to both document passages and structured knowledge graph facts.
Use both sources to answer questions accurately. Cite document sources as [DOC-N] and graph facts as [GRAPH].
If information is not in the provided context, say so clearly.
"""

Microsoft GraphRAG

Microsoft released an open-source GraphRAG implementation that:

1. Extracts a community-detection graph from source documents
2. Creates hierarchical summaries at each community level
3. Retrieves at the right community granularity for the query
4. Synthesizes answers across the community summaries

pip install graphrag

# Index a corpus
python -m graphrag.index --root ./my_project

# Query
python -m graphrag.query \
  --root ./my_project \
  --method global \
  --query "What are the main themes in this document collection?"

GraphRAG’s global search mode synthesizes answers across the entire corpus by using community summaries — useful for questions like “What are the main themes?” that require a bird’s-eye view.

When to Use Graph RAG

Scenario	Standard RAG	Graph RAG
Find relevant passages	✅	✅
Semantic similarity search	✅	✅
Multi-hop entity reasoning	❌	✅
Relationship traversal	❌	✅
Aggregation over entities	❌	✅
Unstructured document corpus	✅	Moderate overhead
Structured domain (medical, legal, finance)	Limited	✅
Small corpus (<1000 docs)	✅	Overkill
Large enterprise knowledge base	✅	✅ (preferred)

Engineering Considerations

Graph construction cost: Entity extraction with an LLM costs time and money. For large corpora, batch extraction with rate limiting and caching is essential.

Graph quality: The graph is only as good as the extraction. Always validate extracted relationships against a sample, and implement confidence thresholds.

Cypher generation reliability: LLM-generated Cypher fails on complex queries. Mitigation: provide schema in the prompt, add retry with error feedback, and maintain a library of validated Cypher templates for common query patterns.

Hybrid search complexity: Managing two retrieval paths adds operational overhead. Start with standard RAG, add graph retrieval only when you have concrete evidence that relational reasoning is needed.

Graph maintenance: When source documents change, the graph must be updated. Build an incremental ingestion pipeline that detects changed documents and updates affected nodes and edges.

---

Incremental Graph Update Pipeline

from pathlib import Path
import hashlib
import json
from neo4j import GraphDatabase

class IncrementalGraphBuilder:
    """Maintains a knowledge graph with incremental document updates."""

    def __init__(self, neo4j_uri: str, user: str, password: str):
        self.driver   = GraphDatabase.driver(neo4j_uri, auth=(user, password))
        self.registry = {}   # doc_path → (hash, timestamp)
        self._load_registry()

    def _load_registry(self):
        with self.driver.session() as session:
            result = session.run(
                "MATCH (d:Document) RETURN d.path as path, d.hash as hash"
            )
            for rec in result:
                self.registry[rec["path"]] = rec["hash"]

    def _doc_hash(self, path: str) -> str:
        return hashlib.md5(Path(path).read_bytes()).hexdigest()

    def needs_update(self, doc_path: str) -> bool:
        current_hash = self._doc_hash(doc_path)
        return self.registry.get(doc_path) != current_hash

    def delete_doc_entities(self, doc_path: str):
        """Remove all entities and relationships from a changed document."""
        with self.driver.session() as session:
            session.run(
                """
                MATCH (e)-[:EXTRACTED_FROM]->(d:Document {path: $path})
                DETACH DELETE e
                """,
                path=doc_path,
            )
            session.run(
                "MATCH (d:Document {path: $path}) DELETE d",
                path=doc_path,
            )

    def ingest_document(self, doc_path: str, llm):
        if not self.needs_update(doc_path):
            return {"status": "skipped", "path": doc_path}

        # Delete stale data
        self.delete_doc_entities(doc_path)

        # Extract text
        text = Path(doc_path).read_text()[:8000]  # first 8k chars

        # Extract graph from new version
        extraction = extract_graph(text, llm)

        # Write to Neo4j with document provenance
        doc_hash = self._doc_hash(doc_path)
        with self.driver.session() as session:
            session.run(
                "MERGE (d:Document {path: $path}) SET d.hash = $hash",
                path=doc_path, hash=doc_hash,
            )
            for rel in extraction.relationships:
                session.run(
                    f"""
                    MERGE (s:{rel.source_type} )
                    MERGE (t:{rel.target_type} )
                    MERGE (s)-[r:{rel.relation.upper().replace(' ', '_')}]->(t)
                    SET r.confidence = $conf
                    MERGE (s)-[:EXTRACTED_FROM]->(:Document )
                    MERGE (t)-[:EXTRACTED_FROM]->(:Document )
                    """,
                    source=rel.source, target=rel.target,
                    conf=rel.confidence, path=doc_path,
                )

        self.registry[doc_path] = doc_hash
        return {"status": "updated", "path": doc_path, "entities": len(extraction.relationships)}

---

Graph Schema Design for Common Domains

Medical knowledge graph

// Medical domain schema
CREATE CONSTRAINT drug_name IF NOT EXISTS FOR (d:Drug) REQUIRE d.name IS UNIQUE;
CREATE CONSTRAINT disease_name IF NOT EXISTS FOR (d:Disease) REQUIRE d.name IS UNIQUE;
CREATE CONSTRAINT protein_name IF NOT EXISTS FOR (p:Protein) REQUIRE p.name IS UNIQUE;

// Example relationships
CREATE (m:Drug {name: "Metformin", mechanism: "AMPK activation"})
CREATE (d:Disease {name: "Type 2 Diabetes", icd10: "E11"})
CREATE (p:Protein {name: "AMPK", function: "energy sensing kinase"})
CREATE (m)-[:TREATS {evidence_level: "A", rct_count: 150}]->(d)
CREATE (m)-[:ACTIVATES {affinity_nm: 0.8}]->(p)
CREATE (p)-[:REGULATES {effect: "inhibits gluconeogenesis"}]->(d)

Legal knowledge graph

// Legal domain schema
CREATE (l:Law {title: "GDPR", jurisdiction: "EU", effective_date: "2018-05-25"})
CREATE (a:Article {number: 17, title: "Right to Erasure"})
CREATE (r:Requirement {text: "Data must be erased upon request"})
CREATE (l)-[:CONTAINS]->(a)
CREATE (a)-[:MANDATES]->(r)
CREATE (r)-[:APPLIES_TO {condition: "when consent withdrawn"}]->(:DataCategory {name: "PersonalData"})

---

Graph RAG with LlamaIndex

LlamaIndex provides a high-level API for Graph RAG with automatic entity extraction:

from llama_index.core import SimpleDirectoryReader, StorageContext
from llama_index.core.indices.knowledge_graph import KnowledgeGraphIndex
from llama_index.llms.openai import OpenAI
from llama_index.core import Settings

Settings.llm      = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.chunk_size = 1024

# Load documents
documents = SimpleDirectoryReader("./data").load_data()

# Build knowledge graph (automatic entity extraction)
storage_context = StorageContext.from_defaults()
kg_index = KnowledgeGraphIndex.from_documents(
    documents,
    storage_context=storage_context,
    max_triplets_per_chunk=10,
    include_embeddings=True,       # enable hybrid retrieval
    show_progress=True,
)

# Query with hybrid retrieval
query_engine = kg_index.as_query_engine(
    include_text=True,             # include source doc text
    retriever_mode="hybrid",       # graph + vector
    response_mode="tree_summarize",
    embedding_mode="hybrid",
    similarity_top_k=5,
)

response = query_engine.query(
    "What drugs interact with metformin in diabetic patients?"
)
print(response)
print("\nSources:")
for node in response.source_nodes:
    print(f"  - {node.metadata.get('file_name')} (score: {node.score:.3f})")

---

Visualizing the Knowledge Graph

# Visualize subgraph with NetworkX + PyVis
import networkx as nx
from pyvis.network import Network
from neo4j import GraphDatabase

def visualize_entity_neighborhood(
    entity_name: str,
    driver: GraphDatabase.driver,
    max_hops: int = 2,
    output_html: str = "graph.html",
):
    G = nx.DiGraph()

    with driver.session() as session:
        result = session.run(
            f"""
            MATCH path = (e )-[*1..{max_hops}]-(neighbor)
            RETURN path
            LIMIT 100
            """,
            name=entity_name,
        )
        for record in result:
            path = record["path"]
            for i, node in enumerate(path.nodes):
                G.add_node(node["name"], label=list(node.labels)[0])
            for rel in path.relationships:
                G.add_edge(
                    rel.start_node["name"],
                    rel.end_node["name"],
                    label=type(rel).__name__,
                )

    # Render with PyVis
    net = Network(height="600px", width="100%", directed=True, notebook=False)
    net.from_nx(G)
    net.toggle_physics(True)
    net.save_graph(output_html)
    print(f"Graph saved to {output_html}")
    return G

---

Benchmarking Graph RAG vs Standard RAG

from dataclasses import dataclass

@dataclass
class EvalCase:
    question:     str
    answer:       str
    query_type:   str   # single-hop, multi-hop, relational, aggregation

EVAL_CASES = [
    EvalCase(
        question="What is the capital of France?",
        answer="Paris",
        query_type="single-hop",
    ),
    EvalCase(
        question="Who is the CEO of the company that acquired Instagram?",
        answer="Mark Zuckerberg",
        query_type="multi-hop",
    ),
    EvalCase(
        question="What drugs treat Type 2 Diabetes AND interact with ACE inhibitors?",
        answer="Metformin, Sulfonylureas (with some caution)",
        query_type="relational",
    ),
    EvalCase(
        question="How many European regulations does GDPR Article 17 reference?",
        answer="3",
        query_type="aggregation",
    ),
]

def compare_systems(standard_rag, graph_rag, eval_cases: list[EvalCase], judge_llm) -> dict:
    results = {"standard_rag": {}, "graph_rag": {}}

    for case in eval_cases:
        std_answer  = standard_rag.query(case.question)
        graph_answer = graph_rag.query(case.question)

        std_score   = judge_llm.score(case.question, case.answer, std_answer)
        graph_score = judge_llm.score(case.question, case.answer, graph_answer)

        results["standard_rag"][case.query_type] = results["standard_rag"].get(case.query_type, []) + [std_score]
        results["graph_rag"][case.query_type]    = results["graph_rag"].get(case.query_type, []) + [graph_score]

    # Aggregate
    comparison = {}
    for qt in set(c.query_type for c in eval_cases):
        std_avg   = sum(results["standard_rag"].get(qt, [0])) / max(len(results["standard_rag"].get(qt, [1])), 1)
        graph_avg = sum(results["graph_rag"].get(qt, [0])) / max(len(results["graph_rag"].get(qt, [1])), 1)
        comparison[qt] = {
            "standard_rag": round(std_avg, 3),
            "graph_rag":    round(graph_avg, 3),
            "winner":       "graph_rag" if graph_avg > std_avg else "standard_rag",
        }

    return comparison

# Expected results:
# single-hop:    standard_rag wins (simple, fast)
# multi-hop:     graph_rag wins (relationship traversal)
# relational:    graph_rag wins significantly
# aggregation:   graph_rag wins (Cypher COUNT/SUM)

---

## LightRAG: Efficient Graph RAG

LightRAG is a simpler, faster alternative to full Neo4j-based Graph RAG that stores a knowledge graph in flat files:

```python
# pip install lightrag-hku

from lightrag import LightRAG, QueryParam
from lightrag.llm import gpt_4o_mini_complete, openai_embedding
import asyncio

WORKING_DIR = "./lightrag_storage"

rag = LightRAG(
    working_dir=WORKING_DIR,
    llm_model_func=gpt_4o_mini_complete,
    embedding_func=openai_embedding,
)

# Insert documents (builds graph + vector index automatically)
with open("medical_papers.txt") as f:
    await rag.ainsert(f.read())

# Query modes
# Naive: standard vector retrieval
naive = await rag.aquery("What drugs treat Type 2 Diabetes?", param=QueryParam(mode="naive"))

# Local: entity-neighborhood retrieval
local = await rag.aquery("What drugs treat Type 2 Diabetes?", param=QueryParam(mode="local"))

# Global: community-level reasoning
global_q = await rag.aquery("What are the main disease categories in the corpus?", param=QueryParam(mode="global"))

# Hybrid: combines all modes
hybrid = await rag.aquery("What drugs treat Type 2 Diabetes?", param=QueryParam(mode="hybrid"))

---

Graph RAG for Enterprise Knowledge Bases

# Real-world: IT infrastructure knowledge graph
from neo4j import GraphDatabase

# Schema for IT knowledge graph
SCHEMA = """
// IT Service Graph Schema
CREATE CONSTRAINT service_name IF NOT EXISTS FOR (s:Service) REQUIRE s.name IS UNIQUE;
CREATE CONSTRAINT host_id IF NOT EXISTS FOR (h:Host) REQUIRE h.id IS UNIQUE;

// Example data
MERGE (web:Service {name: "WebApp", port: 443, status: "running"})
MERGE (db:Service  {name: "PostgreSQL", port: 5432, status: "running"})
MERGE (cache:Service {name: "Redis", port: 6379, status: "running"})
MERGE (host1:Host  {id: "prod-server-01", ip: "10.0.1.10", region: "us-east-1"})
MERGE (host2:Host  {id: "prod-db-01",    ip: "10.0.1.20", region: "us-east-1"})

// Relationships
MERGE (web)-[:DEPENDS_ON {latency_ms: 5}]->(db)
MERGE (web)-[:DEPENDS_ON {latency_ms: 1}]->(cache)
MERGE (web)-[:RUNS_ON]->(host1)
MERGE (db)-[:RUNS_ON]->(host2)
"""

# Query: "What services will be affected if prod-db-01 goes down?"
IMPACT_QUERY = """
MATCH (host:Host {id: $host_id})<-[:RUNS_ON]-(service:Service)
MATCH (dependent:Service)-[:DEPENDS_ON]->(service)
RETURN service.name AS failing_service,
       collect(dependent.name) AS affected_services
"""

def analyze_blast_radius(host_id: str, driver) -> dict:
    with driver.session() as session:
        result = session.run(IMPACT_QUERY, host_id=host_id)
        return [dict(rec) for rec in result]

# Returns: [{'failing_service': 'PostgreSQL', 'affected_services': ['WebApp']}]

---

Community Detection for Global Queries

Microsoft GraphRAG uses community detection to enable global queries across large corpora:

import networkx as nx
from networkx.algorithms.community import louvain_communities

def build_nx_graph_from_neo4j(driver) -> nx.Graph:
    G = nx.Graph()
    with driver.session() as session:
        nodes = session.run("MATCH (n) RETURN id(n) as id, labels(n) as labels, n.name as name")
        for rec in nodes:
            G.add_node(rec["id"], labels=rec["labels"], name=rec["name"])

        edges = session.run("MATCH (a)-[r]->(b) RETURN id(a) as src, id(b) as tgt, type(r) as rel")
        for rec in edges:
            G.add_edge(rec["src"], rec["tgt"], relation=rec["rel"])

    return G

def detect_communities(G: nx.Graph) -> dict[int, list[int]]:
    communities = louvain_communities(G, seed=42)
    return {i: list(comm) for i, comm in enumerate(communities)}

def generate_community_summary(
    community_nodes: list[int],
    G: nx.Graph,
    llm,
) -> str:
    node_names = [G.nodes[n].get("name", str(n)) for n in community_nodes[:20]]
    prompt = f"""Summarize what this cluster of related entities represents:
Entities: {', '.join(node_names)}
Write a 2-3 sentence summary of their common theme."""
    return llm.invoke(prompt).content

---

Conclusion

Graph RAG is a powerful pattern but not always the right tool. The clearest signal that you need it is users asking questions that standard RAG consistently fails to answer correctly — specifically questions involving relationships, multi-hop reasoning, or structured domain knowledge. Start with standard RAG, measure where it fails, and add graph retrieval as a targeted enhancement. The engineering overhead of graph construction, Cypher generation, and graph maintenance is real — but for domains where entity relationships are central to the knowledge (medical, legal, financial, enterprise IT), the quality improvement is substantial and the ROI clear. Tools like LightRAG and Microsoft’s GraphRAG make adoption more accessible than ever, while Neo4j remains the production standard for large-scale, schema-rich knowledge graphs.

---

Graph RAG Technology Stack

Component	Options	Recommended
Graph database	Neo4j, Amazon Neptune, ArangoDB, TigerGraph	Neo4j (best LLM ecosystem)
Entity extraction	GPT-4o, LLM-based custom	GPT-4o + Pydantic
Graph query	Cypher (Neo4j), SPARQL, Gremlin	Cypher (best LLM support)
Vector index	Integrated Neo4j vector, separate Qdrant	Neo4j vector (co-located)
Visualization	Neo4j Browser, NetworkX + PyVis	PyVis for custom dashboards
Managed service	Neo4j AuraDB, AWS Neptune	Neo4j AuraDB (easiest setup)

When to Choose Graph RAG vs. Standard RAG

Question involves entities AND relationships?
  ├── YES: Does it require traversing 2+ hops?
  │     ├── YES: Graph RAG (Cypher traversal)
  │     └── NO:  Hybrid approach (vector + one-hop graph)
  └── NO:  Standard RAG (vector search sufficient)

Question requires counting, aggregating, or ranking entities?
  └── YES: Graph RAG with Cypher aggregation

Domain is medical, legal, or financial knowledge?
  └── YES: Graph RAG strongly recommended

Corpus < 1,000 documents with simple Q&A?
  └── YES: Standard RAG (Graph RAG overhead not worth it)

---

Graph RAG Engineering Principles

1. Start with standard RAG: Only add graph retrieval when you have evidence of relational query failures
2. Entity normalization is critical: “OpenAI”, “Open AI”, and “open_ai” must resolve to the same node
3. Provide graph schema in Cypher prompt: LLMs generate much better Cypher when they know node types and relationship names
4. Always add error-correction retry: LLM-generated Cypher fails on edge cases; retry with error message as context
5. Index graph incrementally: Batch extraction is expensive; use doc hashing to skip unchanged documents
6. Validate extraction quality: Sample 50 extracted triples per domain and verify against source documents
7. Combine with vector retrieval: Graph alone misses semantic nuances; hybrid retrieval covers both cases
8. Monitor Cypher failure rate: If > 20% of graph queries fail, improve the schema description or add validated Cypher templates

Graph RAG is the right tool when entity relationships are central to the domain knowledge. For medical, legal, financial, and enterprise IT knowledge bases, the quality improvement from structured relational reasoning justifies the added complexity of graph construction and maintenance.

Graph RAG Resources

• Microsoft GraphRAG — Open-source community-detection Graph RAG
• LightRAG — Efficient local-first Graph RAG implementation
• Neo4j LLM Integration — LangChain Neo4j integration
• LlamaIndex Knowledge Graph — Automatic graph construction
• Neo4j AuraDB — Managed Neo4j for Graph RAG in production
• Cypher Query Language — Neo4j’s graph query language reference

Graph RAG: Key Takeaways

• Start with standard RAG; only add Graph RAG when you have evidence of relational query failures
• Entity normalization is critical—inconsistent names silently break graph traversal
• Provide the graph schema in every Cypher-generation prompt
• Always add error-correction retry for LLM-generated Cypher—it fails on edge cases
• Index documents incrementally using doc hashing to avoid expensive full re-ingestion
• Validate extraction quality on a sample before scaling graph construction
• Hybrid retrieval (graph + vector) covers both relational and semantic queries
• Graph RAG adds operational overhead; the ROI is highest in structured knowledge domains

---

Graph RAG Conclusion

Graph RAG addresses the fundamental limitation of vector retrieval: it cannot traverse relationships. When users ask questions that require following connections between entities—who manages what, which drugs interact with which targets, what regulations apply to which activities—knowledge graphs provide the structured representation needed to answer them reliably. The engineering investment is real—LLM-based entity extraction, Neo4j infrastructure, Cypher generation, graph maintenance—but for relationship-heavy domains, the quality improvement is substantial and the ROI clear. Start with standard RAG, identify the relational gaps, and add graph retrieval as a targeted enhancement where evidence supports it.

Failure	Symptom	Fix
LLM-generated Cypher fails	Neo4j query error on 30% of queries	Provide schema in prompt; add error-correction retry
Entity extraction misses relationships	Graph is sparse, poor traversal	Use larger model (GPT-4o); add domain examples
Cypher returns no results	Valid query but empty graph match	Check entity normalization (case, typos)
Graph and vector disagree	Conflicting context	Favor graph for relationships, vector for prose evidence
Graph becomes stale	Outdated entities	Implement incremental ingestion with doc hash tracking
Cost too high for graph construction	LLM extraction for 100K+ docs	Use smaller extraction model; cache extraction results

# Cypher query with error correction
def safe_cypher_query(question: str, schema: str, driver, llm, max_retries: int = 2) -> list:
    cypher = generate_cypher(question, schema, llm)
    for attempt in range(max_retries + 1):
        try:
            with driver.session() as session:
                return [dict(r) for r in session.run(cypher)]
        except Exception as e:
            if attempt == max_retries:
                return []   # graceful degradation
            # Ask LLM to fix the error
            cypher = llm.invoke(
                f"Fix this Cypher query:\n{cypher}\nError: {e}\n\nReturn only the corrected query."
            ).content.strip()

# Graph RAG Production Architecture
class ProductionGraphRAG:
    def __init__(self, driver, vectorstore, llm):
        self.driver      = driver
        self.vectorstore = vectorstore
        self.llm         = llm

    def query(self, question: str) -> dict:
        # 1. Classify query type
        qtype = self._classify(question)

        # 2. Dual retrieval
        vector_results = self.vectorstore.similarity_search(question, k=5)
        graph_results  = safe_cypher_query(question, GRAPH_SCHEMA, self.driver, self.llm) if qtype in ("relational", "multi-hop") else []

        # 3. Build context
        context = self._build_context(vector_results, graph_results)

        # 4. Generate grounded answer
        answer = self.llm.invoke(
            f"Answer based on context only.\n\nContext:\n{context}\n\nQuestion: {question}"
        ).content

        return {"answer": answer, "query_type": qtype, "graph_facts": len(graph_results)}

    def _classify(self, question: str) -> str:
        prompt = f"Classify as: semantic, relational, multi-hop, or aggregation.\nQuestion: {question}\nAnswer (one word):"
        return self.llm.invoke(prompt).content.strip().lower()

    def _build_context(self, vector_docs: list, graph_facts: list) -> str:
        parts = []
        if vector_docs:
            parts.append("## Document Evidence\n" + "\n\n".join(d.page_content for d in vector_docs))
        if graph_facts:
            parts.append("## Knowledge Graph Facts\n" + "\n".join(str(f) for f in graph_facts))
        return "\n\n".join(parts)

---

Production Engineering Notes

Building production LLM systems requires continuous learning and adaptation. The patterns and tools in this ecosystem evolve rapidly, but certain engineering principles remain constant:

On reliability: The most reliable systems are built with defense in depth � multiple independent validation layers, graceful degradation paths, circuit breakers, and comprehensive observability. No single component should be a single point of failure.

On evaluation: Automated evaluation enables speed; human evaluation provides ground truth. Calibrate your automated judges regularly against human labels. A 10-point quality regression detected in CI costs hours to fix; the same regression discovered after deployment costs days or weeks of user trust.

On iteration: LLM systems improve through measurement, not intuition. Every prompt change, model upgrade, and retrieval configuration should be evaluated against a representative dataset before deployment. Blind experimentation is expensive; instrumented experimentation is invaluable.

On operations: Monitor cost, latency, and quality as a unified picture. A system that is fast and cheap but inaccurate is not production-grade. A system that is accurate but 10x over budget is not sustainable. The engineering goal is acceptable quality at acceptable cost within acceptable latency.