Graph RAG: Building Knowledge Graph-Enhanced Retrieval Pipelines for Complex Multi-Hop Queries

Standard RAG fails in a predictable way. You embed a question, fetch the top-k chunks by cosine similarity, and stuff them into a prompt. For a question like "What are the side effects of Drug X?" that works fine. But ask "What are the contraindications for Drug X given that the patient is also taking Drug Y, which was prescribed for a condition that affects the same metabolic pathway?"—and your vector search returns chunks that are each individually relevant but collectively incomplete. The reasoning that connects them is nowhere in the retrieved context.

This is the multi-hop problem. The answer requires traversing a chain of relationships: Drug X → metabolic pathway → Drug Y → contraindication. Vector similarity finds islands of relevance; it doesn't find the bridges between them. Knowledge graphs do. Graph RAG is the architectural pattern that combines the associative recall of vector search with the structured traversal capabilities of a knowledge graph, and it's one of the more powerful techniques available for building AI systems that need to reason over complex, interconnected domains.

By the end of this lesson, you'll have built a working Graph RAG pipeline from scratch. You'll understand not just the mechanics but the why behind each design decision—because Graph RAG has a lot of moving parts, and if you don't understand why each piece exists, you'll make the wrong trade-offs when the inevitable edge cases appear.

What you'll learn:

• How knowledge graphs complement (and sometimes replace) vector stores in retrieval pipelines
• How to construct a knowledge graph from unstructured text using entity and relationship extraction
• How to implement hybrid retrieval that uses vector search as an entry point and graph traversal for expansion
• How to translate natural language queries into graph traversal strategies, including Cypher generation via LLM
• How to design Graph RAG pipelines that scale—and where they break down

---

Prerequisites

You should be comfortable with:

• Standard RAG pipelines (embeddings, vector stores, chunking strategies)
• Python at a level where you can wire together APIs and data pipelines
• Basic graph theory (nodes, edges, directed vs. undirected graphs)
• At least passing familiarity with a graph database (Neo4j is what we'll use here)

You do not need to be a Neo4j expert. We'll cover the Cypher queries you need as we go.

---

Why Vector Search Alone Isn't Enough for Complex Queries

Before we build anything, let's be precise about the failure mode. Consider a corpus of pharmaceutical research documents. You've embedded everything and built a solid vector index. A user asks:

"Which drugs prescribed for Type 2 diabetes interact with the ACE inhibitors commonly used in patients who also have chronic kidney disease?"

The information needed to answer this question is distributed across at least four conceptual clusters:

1. Drugs used for Type 2 diabetes
2. ACE inhibitors and their typical use cases
3. The overlap between Type 2 diabetes and chronic kidney disease populations
4. Drug interaction mechanisms between those two drug classes

A vector search for the full question will likely surface chunks about diabetes drugs, maybe some about ACE inhibitors, and possibly some about CKD. But the relationships—"drug X treats condition Y," "drug X interacts with drug class Z"—are implicit in the text. They're scattered across paragraphs. The retriever can't assemble them into a coherent chain.

What you really need is something like:

(Drug)-[:TREATS]->(Type2Diabetes)
(Drug)-[:BELONGS_TO]->(ACEInhibitorClass)
(Type2Diabetes)<-[:COMORBID_WITH]-(ChronicKidneyDisease)
(Drug)-[:INTERACTS_WITH]->(Drug)

Then you traverse: Give me all (Drug A)-[:TREATS]->(Type2Diabetes) nodes, find all (Drug B)-[:INDICATED_FOR]->(ChronicKidneyDisease), check if any (Drug A)-[:INTERACTS_WITH]->(Drug B) edges exist.

That's a three-hop traversal. Vector search cannot do this natively. Graph databases are built for exactly this.

The key insight: vector search excels at finding semantically similar content; graph traversal excels at following typed relationships across entities. Graph RAG combines both. Vector search finds your starting nodes; graph traversal expands the neighborhood.

---

The Architecture of a Graph RAG Pipeline

Here's the high-level architecture we'll be building:

                        ┌─────────────────────────────────────┐
                        │           INGESTION PIPELINE         │
                        │                                      │
  Raw Documents ──────► │  1. Chunk & Extract Entities/Rels   │
                        │  2. Embed entities + chunk text      │
                        │  3. Write to Neo4j + Vector Store    │
                        └─────────────────────────────────────┘
                                           │
                                           ▼
                        ┌─────────────────────────────────────┐
                        │            KNOWLEDGE GRAPH           │
                        │   (Neo4j)                            │
                        │   Nodes: Entities + Document Chunks  │
                        │   Edges: Typed relationships         │
                        │   + Vector index on node embeddings  │
                        └─────────────────────────────────────┘
                                           │
                                           ▼
                        ┌─────────────────────────────────────┐
                        │           QUERY PIPELINE             │
                        │                                      │
  User Query ─────────► │  1. Entity extraction from query    │
                        │  2. Vector search → seed nodes       │
                        │  3. Graph traversal → expansion      │
                        │  4. Context assembly                 │
                        │  5. LLM generation                   │
                        └─────────────────────────────────────┘

There are two pipelines: ingestion and query. Most tutorials focus on the query pipeline because it's more interesting, but the ingestion pipeline is where most real-world Graph RAG projects succeed or fail. Garbage graph, garbage answers.

Let's build both.

---

Setting Up Neo4j and the Environment

We'll use Neo4j (community edition is fine) with the APOC and GDS plugins. Neo4j has a built-in vector index capability as of version 5.x, which eliminates the need for a separate vector store in many cases.

# requirements
# pip install neo4j openai langchain langchain-openai 
# pip install spacy sentence-transformers

import os
from neo4j import GraphDatabase
from openai import OpenAI
from sentence_transformers import SentenceTransformer
import json
import re
from typing import Optional

# Configuration
NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687")
NEO4J_USER = os.getenv("NEO4J_USER", "neo4j")
NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "your_password")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

client = OpenAI(api_key=OPENAI_API_KEY)
embedder = SentenceTransformer("all-MiniLM-L6-v2")  # Fast, 384-dim

class GraphRAGPipeline:
    def __init__(self):
        self.driver = GraphDatabase.driver(
            NEO4J_URI, 
            auth=(NEO4J_USER, NEO4J_PASSWORD)
        )
        self._setup_schema()
    
    def _setup_schema(self):
        """Create indexes and constraints."""
        with self.driver.session() as session:
            # Uniqueness constraints
            session.run("""
                CREATE CONSTRAINT entity_id IF NOT EXISTS
                FOR (e:Entity) REQUIRE e.id IS UNIQUE
            """)
            session.run("""
                CREATE CONSTRAINT chunk_id IF NOT EXISTS
                FOR (c:Chunk) REQUIRE c.id IS UNIQUE
            """)
            # Vector index for semantic search on chunks
            session.run("""
                CREATE VECTOR INDEX chunk_embeddings IF NOT EXISTS
                FOR (c:Chunk) ON (c.embedding)
                OPTIONS {indexConfig: {
                    `vector.dimensions`: 384,
                    `vector.similarity_function`: 'cosine'
                }}
            """)
            # Vector index for entity search
            session.run("""
                CREATE VECTOR INDEX entity_embeddings IF NOT EXISTS
                FOR (e:Entity) ON (e.embedding)
                OPTIONS {indexConfig: {
                    `vector.dimensions`: 384,
                    `vector.similarity_function`: 'cosine'
                }}
            """)
    
    def close(self):
        self.driver.close()

The schema design deserves explanation. We have two node types: Entity (the extracted knowledge graph nodes—drugs, conditions, proteins, companies, whatever fits your domain) and Chunk (the actual text passages from source documents). Entities and Chunks are connected by a MENTIONED_IN relationship, and entities are connected to each other by domain-specific relationship types. This dual structure lets us do vector search over raw text and over entities separately, then traverse between them.

---

Building the Ingestion Pipeline: Entity and Relationship Extraction

This is the hardest part of Graph RAG and the part most tutorials skip. You need to turn unstructured text into a typed, consistent graph. The primary tool is an LLM with structured output.

EXTRACTION_PROMPT = """You are a knowledge graph extraction assistant. 
Given the following text passage, extract all entities and relationships.

Return a JSON object with this exact structure:
{
  "entities": [
    {
      "id": "unique_snake_case_identifier",
      "name": "Entity Name",
      "type": "DRUG|CONDITION|PROTEIN|GENE|PATHWAY|ORGANIZATION|PERSON|CONCEPT",
      "description": "brief description"
    }
  ],
  "relationships": [
    {
      "source": "source_entity_id",
      "target": "target_entity_id", 
      "type": "RELATIONSHIP_TYPE_IN_CAPS",
      "properties": {"confidence": 0.9, "direction": "source_to_target"}
    }
  ]
}

Use consistent entity IDs across the entire extraction — if "metformin" appears 
multiple times, always use id "metformin". Use specific relationship types that 
carry semantic meaning: TREATS, INHIBITS, ACTIVATES, CAUSES, CONTRAINDICATED_WITH,
METABOLIZED_BY, ENCODED_BY, etc.

Text passage:
{text}

Return only valid JSON, no other text."""

def extract_graph_elements(text: str, chunk_id: str) -> dict:
    """Extract entities and relationships from a text chunk."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "user", "content": EXTRACTION_PROMPT.format(text=text)}
        ],
        response_format={"type": "json_object"},
        temperature=0  # Deterministic extraction
    )
    
    try:
        result = json.loads(response.choices[0].message.content)
        # Attach source chunk reference to all entities
        for entity in result.get("entities", []):
            entity["source_chunk"] = chunk_id
        return result
    except json.JSONDecodeError as e:
        print(f"Failed to parse extraction for chunk {chunk_id}: {e}")
        return {"entities": [], "relationships": []}

Warning: Entity resolution is your biggest quality risk. The LLM might extract "Metformin," "metformin HCl," and "metformin hydrochloride" as three separate entities. In your graph, these need to be the same node. Plan for a normalization/deduplication step. The simplest approach is to lowercase and strip common suffixes before generating IDs; a more robust approach uses embedding similarity clustering on entity names before loading.

def normalize_entity_id(name: str) -> str:
    """Basic entity ID normalization. Extend with domain-specific rules."""
    normalized = name.lower().strip()
    normalized = re.sub(r'\s+', '_', normalized)
    normalized = re.sub(r'[^a-z0-9_]', '', normalized)
    # Strip common pharmaceutical suffixes
    for suffix in ['_hydrochloride', '_hcl', '_sodium', '_potassium']:
        if normalized.endswith(suffix):
            normalized = normalized[:-len(suffix)]
    return normalized

def ingest_document(pipeline: GraphRAGPipeline, text: str, doc_id: str, 
                    chunk_size: int = 512, overlap: int = 64):
    """Full ingestion pipeline for a single document."""
    
    # 1. Chunk the document
    chunks = create_overlapping_chunks(text, chunk_size, overlap)
    
    all_entities = {}
    all_relationships = []
    
    for i, chunk_text in enumerate(chunks):
        chunk_id = f"{doc_id}_chunk_{i}"
        
        # 2. Store the chunk with its embedding
        chunk_embedding = embedder.encode(chunk_text).tolist()
        
        with pipeline.driver.session() as session:
            session.run("""
                MERGE (c:Chunk {id: $chunk_id})
                SET c.text = $text,
                    c.doc_id = $doc_id,
                    c.chunk_index = $index,
                    c.embedding = $embedding
            """, chunk_id=chunk_id, text=chunk_text, 
                doc_id=doc_id, index=i, embedding=chunk_embedding)
        
        # 3. Extract entities and relationships
        extracted = extract_graph_elements(chunk_text, chunk_id)
        
        for entity in extracted["entities"]:
            normalized_id = normalize_entity_id(entity["id"])
            entity["id"] = normalized_id
            
            if normalized_id not in all_entities:
                all_entities[normalized_id] = entity
                all_entities[normalized_id]["source_chunks"] = []
            
            all_entities[normalized_id]["source_chunks"].append(chunk_id)
        
        all_relationships.extend(extracted["relationships"])
    
    # 4. Write entities to graph
    _write_entities(pipeline, all_entities)
    
    # 5. Write relationships to graph
    _write_relationships(pipeline, all_relationships)
    
    # 6. Link entities to their source chunks
    _link_entities_to_chunks(pipeline, all_entities)
    
    print(f"Ingested {doc_id}: {len(chunks)} chunks, "
          f"{len(all_entities)} entities, {len(all_relationships)} relationships")

def _write_entities(pipeline: GraphRAGPipeline, entities: dict):
    with pipeline.driver.session() as session:
        for entity_id, entity in entities.items():
            name_embedding = embedder.encode(
                f"{entity['name']}: {entity.get('description', '')}"
            ).tolist()
            
            session.run("""
                MERGE (e:Entity {id: $id})
                SET e.name = $name,
                    e.type = $type,
                    e.description = $description,
                    e.embedding = $embedding
                WITH e
                CALL apoc.create.addLabels(e, [$type]) YIELD node
                RETURN node
            """, id=entity_id, name=entity["name"], 
                type=entity.get("type", "CONCEPT"),
                description=entity.get("description", ""),
                embedding=name_embedding)

def _write_relationships(pipeline: GraphRAGPipeline, relationships: list):
    with pipeline.driver.session() as session:
        for rel in relationships:
            source_id = normalize_entity_id(rel["source"])
            target_id = normalize_entity_id(rel["target"])
            rel_type = rel["type"].upper().replace(" ", "_")
            confidence = rel.get("properties", {}).get("confidence", 0.7)
            
            # APOC's dynamic relationship creation
            session.run("""
                MATCH (s:Entity {id: $source_id})
                MATCH (t:Entity {id: $target_id})
                CALL apoc.create.relationship(s, $rel_type, 
                    {confidence: $confidence}, t) YIELD rel
                RETURN rel
            """, source_id=source_id, target_id=target_id,
                rel_type=rel_type, confidence=confidence)

def _link_entities_to_chunks(pipeline: GraphRAGPipeline, entities: dict):
    with pipeline.driver.session() as session:
        for entity_id, entity in entities.items():
            for chunk_id in entity.get("source_chunks", []):
                session.run("""
                    MATCH (e:Entity {id: $entity_id})
                    MATCH (c:Chunk {id: $chunk_id})
                    MERGE (e)-[:MENTIONED_IN]->(c)
                """, entity_id=entity_id, chunk_id=chunk_id)

---

The Query Pipeline: Hybrid Retrieval with Graph Traversal

Now for the part that makes Graph RAG actually work. The query pipeline has four stages: entity extraction, vector seed search, graph expansion, and context assembly.

def query_graph_rag(pipeline: GraphRAGPipeline, query: str, 
                    max_hops: int = 2, top_k_seeds: int = 5) -> str:
    """Full Graph RAG query pipeline."""
    
    # Stage 1: Extract entities from the query
    query_entities = extract_query_entities(query)
    print(f"Query entities: {query_entities}")
    
    # Stage 2: Find seed nodes via vector search
    seed_nodes = find_seed_nodes(pipeline, query, query_entities, top_k=top_k_seeds)
    print(f"Seed nodes: {[n['id'] for n in seed_nodes]}")
    
    # Stage 3: Expand via graph traversal
    expanded_subgraph = expand_neighborhood(pipeline, seed_nodes, max_hops=max_hops)
    
    # Stage 4: Retrieve relevant chunks for context
    context_chunks = retrieve_context_chunks(pipeline, expanded_subgraph, query)
    
    # Stage 5: Generate answer
    return generate_answer(query, context_chunks, expanded_subgraph)

Stage 1: Query Entity Extraction

def extract_query_entities(query: str) -> list[dict]:
    """Extract entities from the user query to guide graph traversal."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Cheap model is fine here
        messages=[{
            "role": "user",
            "content": f"""Extract named entities from this query. Return JSON:
{{"entities": [{{"name": "entity name", "type": "DRUG|CONDITION|etc"}}]}}

Query: {query}

Return only JSON."""
        }],
        response_format={"type": "json_object"},
        temperature=0
    )
    result = json.loads(response.choices[0].message.content)
    return result.get("entities", [])

Stage 2: Vector Seed Search

def find_seed_nodes(pipeline: GraphRAGPipeline, query: str, 
                    query_entities: list, top_k: int = 5) -> list[dict]:
    """Find starting points in the graph using vector similarity."""
    query_embedding = embedder.encode(query).tolist()
    
    seed_nodes = []
    
    with pipeline.driver.session() as session:
        # Search entity nodes by embedding similarity
        entity_results = session.run("""
            CALL db.index.vector.queryNodes(
                'entity_embeddings', $top_k, $embedding
            ) YIELD node, score
            WHERE score > 0.6
            RETURN node.id as id, node.name as name, 
                   node.type as type, node.description as description,
                   score
            ORDER BY score DESC
        """, top_k=top_k, embedding=query_embedding)
        
        for record in entity_results:
            seed_nodes.append({
                "id": record["id"],
                "name": record["name"],
                "type": record["type"],
                "description": record["description"],
                "score": record["score"],
                "source": "vector_entity_search"
            })
        
        # Also do exact name matching for entities extracted from query
        for entity in query_entities:
            exact_results = session.run("""
                MATCH (e:Entity)
                WHERE toLower(e.name) = toLower($name)
                   OR toLower(e.id) = toLower($normalized_name)
                RETURN e.id as id, e.name as name, 
                       e.type as type, e.description as description,
                       1.0 as score
            """, name=entity["name"], 
                normalized_name=normalize_entity_id(entity["name"]))
            
            for record in exact_results:
                node = {
                    "id": record["id"],
                    "name": record["name"],
                    "type": record["type"],
                    "description": record["description"],
                    "score": record["score"],
                    "source": "exact_match"
                }
                # Avoid duplicates
                if not any(n["id"] == node["id"] for n in seed_nodes):
                    seed_nodes.append(node)
    
    return seed_nodes

Stage 3: Graph Traversal Expansion

This is the core of Graph RAG. Given seed nodes, we traverse the graph to find connected entities and the relationships between them.

def expand_neighborhood(pipeline: GraphRAGPipeline, 
                        seed_nodes: list[dict], 
                        max_hops: int = 2) -> dict:
    """
    Expand from seed nodes using graph traversal.
    Returns a subgraph dict with nodes and edges.
    """
    if not seed_nodes:
        return {"nodes": [], "relationships": []}
    
    seed_ids = [n["id"] for n in seed_nodes]
    
    with pipeline.driver.session() as session:
        # Variable-length path traversal with hop limit
        # We use apoc.path.subgraphAll for controlled expansion
        result = session.run("""
            MATCH (seed:Entity)
            WHERE seed.id IN $seed_ids
            CALL apoc.path.subgraphAll(seed, {
                maxLevel: $max_hops,
                relationshipFilter: ">",
                labelFilter: "+Entity"
            }) YIELD nodes, relationships
            UNWIND nodes AS node
            WITH COLLECT(DISTINCT {
                id: node.id,
                name: node.name,
                type: node.type,
                description: node.description
            }) AS all_nodes, relationships
            UNWIND relationships AS rel
            WITH all_nodes, COLLECT(DISTINCT {
                source: startNode(rel).id,
                target: endNode(rel).id,
                type: type(rel),
                confidence: rel.confidence
            }) AS all_rels
            RETURN all_nodes, all_rels
        """, seed_ids=seed_ids, max_hops=max_hops)
        
        records = list(result)
        if not records:
            return {"nodes": list(seed_nodes), "relationships": []}
        
        record = records[0]
        return {
            "nodes": record["all_nodes"],
            "relationships": record["all_rels"]
        }

Performance note: apoc.path.subgraphAll with max_hops=3 on a large graph can be catastrophically slow if your seed nodes are highly connected hubs. A drug like "aspirin" might be connected to hundreds of conditions. You need degree-aware traversal: cap expansion at nodes with degree > N, or filter by relationship type and confidence threshold. We'll cover this in the optimization section.

Stage 4: Context Chunk Retrieval

Now that we have an expanded subgraph, we retrieve the actual text chunks that contain the relevant entities:

def retrieve_context_chunks(pipeline: GraphRAGPipeline, 
                             subgraph: dict, 
                             query: str,
                             max_chunks: int = 8) -> list[str]:
    """
    Retrieve text chunks for entities in the subgraph,
    ranked by relevance to the original query.
    """
    node_ids = [n["id"] for n in subgraph.get("nodes", [])]
    if not node_ids:
        return []
    
    query_embedding = embedder.encode(query).tolist()
    
    with pipeline.driver.session() as session:
        # Get chunks that mention entities in our subgraph,
        # ranked by embedding similarity to the query
        result = session.run("""
            MATCH (e:Entity)-[:MENTIONED_IN]->(c:Chunk)
            WHERE e.id IN $node_ids
            WITH c, COUNT(DISTINCT e) as entity_coverage
            WITH c, entity_coverage,
                 vector.similarity.cosine(c.embedding, $query_embedding) as chunk_score
            WHERE chunk_score > 0.5
            RETURN c.text as text, 
                   c.id as chunk_id,
                   chunk_score,
                   entity_coverage,
                   (chunk_score * 0.7 + toFloat(entity_coverage) * 0.3) as combined_score
            ORDER BY combined_score DESC
            LIMIT $max_chunks
        """, node_ids=node_ids, query_embedding=query_embedding, 
            max_chunks=max_chunks)
        
        chunks = [record["text"] for record in result]
    
    return chunks

The combined_score formula here is intentional: we blend semantic similarity (how well does this chunk match the query?) with entity coverage (how many of our subgraph entities appear in this chunk?). A chunk that mentions three entities from our traversal is likely to contain relationship information, even if its raw embedding similarity is moderate.

Stage 5: Answer Generation with Graph-Aware Context

def build_graph_context_string(subgraph: dict) -> str:
    """Convert subgraph structure to a readable summary for the LLM."""
    if not subgraph["relationships"]:
        return "No relationships found."
    
    lines = ["**Knowledge Graph Context:**"]
    
    # Group relationships by type
    rel_by_type = {}
    for rel in subgraph["relationships"]:
        rel_type = rel["type"]
        if rel_type not in rel_by_type:
            rel_by_type[rel_type] = []
        rel_by_type[rel_type].append(rel)
    
    # Create entity name lookup
    entity_names = {n["id"]: n["name"] for n in subgraph["nodes"]}
    
    for rel_type, rels in rel_by_type.items():
        lines.append(f"\n{rel_type}:")
        for rel in rels[:10]:  # Cap per relationship type
            source_name = entity_names.get(rel["source"], rel["source"])
            target_name = entity_names.get(rel["target"], rel["target"])
            confidence = rel.get("confidence", 0)
            lines.append(f"  • {source_name} → {target_name} (confidence: {confidence:.2f})")
    
    return "\n".join(lines)

def generate_answer(query: str, context_chunks: list[str], 
                    subgraph: dict) -> str:
    """Generate a final answer using both text chunks and graph structure."""
    
    graph_context = build_graph_context_string(subgraph)
    text_context = "\n\n---\n\n".join(context_chunks)
    
    system_prompt = """You are a precise analytical assistant. You have been given 
two types of context:
1. A knowledge graph showing structured relationships between entities
2. Text passages from source documents

Use BOTH types of context to answer the question. The knowledge graph helps you 
understand multi-hop relationships; the text passages provide detailed evidence.
When citing relationships from the graph, note them explicitly. If information 
is absent from both contexts, say so—do not fabricate."""
    
    user_prompt = f"""Question: {query}

{graph_context}

**Source Document Passages:**
{text_context}

Please provide a comprehensive answer that explicitly traces the reasoning chain 
across the relationships in the knowledge graph."""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0.1
    )
    
    return response.choices[0].message.content

---

Advanced Pattern: LLM-to-Cypher Translation

For power users, you can let the LLM write Cypher directly against your schema. This is powerful but requires careful guardrailing.

CYPHER_GENERATION_PROMPT = """You are a Cypher query generator for a pharmaceutical 
knowledge graph.

The graph has this schema:
- Nodes: (:Entity {id, name, type, description}) with subtypes DRUG, CONDITION, 
  PROTEIN, GENE, PATHWAY
- Edges: TREATS, INHIBITS, ACTIVATES, CAUSES, CONTRAINDICATED_WITH, 
  METABOLIZED_BY, INTERACTS_WITH, ENCODED_BY, PART_OF
- (:Chunk {id, text, doc_id}) connected to entities via (:Entity)-[:MENTIONED_IN]->(:Chunk)

Generate a READ-ONLY Cypher query to answer this question. 
Return ONLY the Cypher, no explanation.
Limit results to 20 rows maximum.
Never use DELETE, MERGE, SET, or CREATE.

Question: {question}"""

def generate_and_execute_cypher(pipeline: GraphRAGPipeline, 
                                 question: str) -> list[dict]:
    """Generate Cypher from natural language and execute it."""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": CYPHER_GENERATION_PROMPT.format(question=question)
        }],
        temperature=0
    )
    
    cypher = response.choices[0].message.content.strip()
    cypher = cypher.replace("```cypher", "").replace("```", "").strip()
    
    # Safety check - rudimentary but important
    forbidden_keywords = ["DELETE", "MERGE", "CREATE", "SET", "REMOVE", "DROP"]
    if any(keyword in cypher.upper() for keyword in forbidden_keywords):
        raise ValueError(f"Generated Cypher contains forbidden operations: {cypher}")
    
    print(f"Generated Cypher:\n{cypher}\n")
    
    with pipeline.driver.session() as session:
        result = session.run(cypher)
        return [dict(record) for record in result]

Security critical: Never execute LLM-generated Cypher without a read-only Neo4j user. Create a dedicated user in Neo4j with only READ access for your application's query pathway. The keyword filter above is a last line of defense, not a primary one. Credential-level access control is mandatory.

---

Scaling Considerations and Where Graph RAG Breaks

Let's be honest about the failure modes, because understanding them lets you design around them.

Failure Mode 1: Hub Node Explosion

In any real-world knowledge graph, some nodes are extremely high-degree. In a pharmaceutical graph, "liver" as a metabolism organ might be connected to thousands of drugs. When "liver" shows up as a seed node or gets traversed to in hop 1, your subgraphAll call tries to load all those connections.

Fix: Degree-filtered traversal

def expand_neighborhood_safe(pipeline: GraphRAGPipeline,
                              seed_ids: list[str],
                              max_hops: int = 2,
                              max_degree: int = 50,
                              min_confidence: float = 0.7) -> dict:
    """Traversal with degree and confidence filtering."""
    
    with pipeline.driver.session() as session:
        result = session.run("""
            MATCH (seed:Entity)
            WHERE seed.id IN $seed_ids
            
            // First hop with degree check
            OPTIONAL MATCH (seed)-[r1]->(hop1:Entity)
            WHERE r1.confidence >= $min_confidence
            WITH seed, hop1, r1,
                 COUNT { (hop1)-[]->() } as hop1_degree
            WHERE hop1_degree <= $max_degree OR hop1 IS NULL
            
            // Second hop with degree check  
            OPTIONAL MATCH (hop1)-[r2]->(hop2:Entity)
            WHERE r2.confidence >= $min_confidence
              AND $max_hops >= 2
            WITH seed, hop1, r1, hop2, r2,
                 COUNT { (hop2)-[]->() } as hop2_degree
            WHERE hop2_degree <= $max_degree OR hop2 IS NULL
            
            // Collect results
            WITH COLLECT(DISTINCT {id: seed.id, name: seed.name, 
                                    type: seed.type}) +
                 COLLECT(DISTINCT CASE WHEN hop1 IS NOT NULL 
                         THEN {id: hop1.id, name: hop1.name, 
                               type: hop1.type} END) +
                 COLLECT(DISTINCT CASE WHEN hop2 IS NOT NULL 
                         THEN {id: hop2.id, name: hop2.name, 
                               type: hop2.type} END) AS all_nodes,
                 COLLECT(DISTINCT CASE WHEN r1 IS NOT NULL
                         THEN {source: seed.id, target: hop1.id,
                               type: type(r1), 
                               confidence: r1.confidence} END) +
                 COLLECT(DISTINCT CASE WHEN r2 IS NOT NULL
                         THEN {source: hop1.id, target: hop2.id,
                               type: type(r2),
                               confidence: r2.confidence} END) AS all_rels
            
            RETURN [n IN all_nodes WHERE n IS NOT NULL] AS nodes,
                   [r IN all_rels WHERE r IS NOT NULL] AS relationships
        """, seed_ids=seed_ids, max_hops=max_hops,
            max_degree=max_degree, min_confidence=min_confidence)
        
        records = list(result)
        if not records:
            return {"nodes": [], "relationships": []}
        
        return {
            "nodes": records[0]["nodes"],
            "relationships": records[0]["relationships"]
        }

Failure Mode 2: Entity Resolution Drift

Over time, as you ingest more documents, your normalization logic may produce inconsistent entity IDs. "ckd" and "chronic_kidney_disease" should be the same node.

Fix: Post-ingestion entity merging using embedding similarity

def merge_similar_entities(pipeline: GraphRAGPipeline, 
                            similarity_threshold: float = 0.92):
    """Find and merge entities that are likely the same."""
    
    with pipeline.driver.session() as session:
        # Find entity pairs with high embedding similarity
        candidates = session.run("""
            CALL db.index.vector.queryNodes('entity_embeddings', 10, e.embedding)
            YIELD node as candidate, score
            WHERE score > $threshold AND candidate.id <> e.id
                  AND candidate.type = e.type
            MATCH (e:Entity)
            RETURN e.id as source_id, candidate.id as target_id, score
            LIMIT 100
        """, threshold=similarity_threshold)  # Note: this is illustrative
        
        # In practice, use APOC's apoc.ml.similarity or a batch approach
        # The actual implementation requires iterating all entity pairs
        # which is better done via the Python driver with batching
        
        for record in candidates:
            print(f"Merge candidate: {record['source_id']} ≈ {record['target_id']} "
                  f"(score: {record['score']:.3f})")
            # Decision: keep the entity with more connections
            session.run("""
                MATCH (keep:Entity {id: $keep_id})
                MATCH (merge:Entity {id: $merge_id})
                WITH keep, merge
                CALL apoc.refactor.mergeNodes([keep, merge], {
                    properties: 'discard',
                    mergeRels: true
                }) YIELD node
                RETURN node
            """, keep_id=record["source_id"], merge_id=record["target_id"])

Failure Mode 3: Relationship Type Inconsistency

The LLM might use "TREATS" in one extraction, "USED_FOR" in another, and "INDICATED_FOR" in a third—for the same semantic relationship. This fragments your graph.

Fix: Define a canonical relationship ontology and enforce it at extraction time. Update your extraction prompt with a strict list of allowed relationship types and add a validation step:

ALLOWED_RELATIONSHIPS = {
    "TREATS", "INHIBITS", "ACTIVATES", "CAUSES", 
    "CONTRAINDICATED_WITH", "METABOLIZED_BY", "INTERACTS_WITH",
    "ENCODED_BY", "PART_OF", "ASSOCIATED_WITH", "REGULATES",
    "EXPRESSED_IN", "BELONGS_TO_CLASS", "SIDE_EFFECT_OF"
}

def validate_relationship_type(rel_type: str) -> str:
    """Map extracted relationship type to canonical ontology."""
    # Exact match
    if rel_type.upper() in ALLOWED_RELATIONSHIPS:
        return rel_type.upper()
    
    # Fuzzy mapping for common variants
    synonyms = {
        "USED_FOR": "TREATS",
        "INDICATED_FOR": "TREATS",
        "PRESCRIBED_FOR": "TREATS",
        "SUPPRESSES": "INHIBITS",
        "BLOCKS": "INHIBITS",
        "INTERACTS": "INTERACTS_WITH",
        "DRUG_DRUG_INTERACTION": "INTERACTS_WITH",
    }
    
    normalized = rel_type.upper().replace(" ", "_")
    return synonyms.get(normalized, "ASSOCIATED_WITH")  # Fallback

---

Hands-On Exercise

You now have all the components to build a working Graph RAG system. Here's a structured exercise to cement the concepts.

Scenario: Build a Graph RAG system over a dataset of clinical trial summaries. The system should answer complex multi-hop queries like "Which drugs currently in Phase 3 trials for treatment-resistant depression have mechanisms that overlap with existing SSRIs?"

Step 1: Data Collection Download 20-30 clinical trial summary texts from ClinicalTrials.gov for depression or any condition of interest. Store them as plain text files.

Step 2: Ingestion Using the ingestion pipeline above, process each document. After ingestion, open the Neo4j Browser (usually at http://localhost:7474) and run:

MATCH (n:Entity) RETURN n.type, COUNT(*) as count ORDER BY count DESC

You should see a distribution of entity types. If you have fewer than 50 unique entities across 20 documents, your extraction prompt is too conservative—adjust it.

Step 3: Graph Inspection Run this query in Neo4j Browser to visualize a subgraph:

MATCH path = (d:DRUG)-[r]->(c:CONDITION)
WHERE type(r) IN ['TREATS', 'INDICATED_FOR']
RETURN path LIMIT 25

If the graph looks right, move on. If you see isolated nodes and no relationships, your relationship extraction is broken—recheck the LLM output for a sample chunk.

Step 4: Query Testing Test with three queries of increasing complexity:

1. Single-hop: "What drugs are used for depression?" (Should work with basic vector search)
2. Two-hop: "What drugs for depression affect serotonin pathways?" (Requires drug → mechanism → pathway traversal)
3. Three-hop: "What drugs for depression have mechanisms similar to SSRIs but are in early-stage trials?" (drug → mechanism → class comparison → trial phase)

Track which queries the standard vector RAG would fail and which Graph RAG handles correctly.

Step 5: Evaluate For each query, compare the answer from:

• Pure vector RAG (top-5 chunk retrieval, no graph)
• Graph RAG with max_hops=1
• Graph RAG with max_hops=2

Note which contexts are richer and which answers are more complete.

---

Common Mistakes & Troubleshooting

"My graph has thousands of nodes but the traversal always returns nothing."

Check your relationship types. In Neo4j Browser, run:

MATCH ()-[r]->() RETURN type(r), COUNT(*) ORDER BY COUNT(*) DESC LIMIT 20

If you see relationship types that don't match what your traversal query expects, you have a naming mismatch. The APOC dynamic relationship creation may have stored "Treats" instead of "TREATS." Fix with a migration:

MATCH ()-[r:Treats]->() 
CALL apoc.refactor.setType(r, 'TREATS') YIELD output RETURN COUNT(output)

"Vector search returns no seed nodes above the similarity threshold."

Your embedder's vector space might not match your chunk content style. The all-MiniLM-L6-v2 model is general purpose; for scientific or legal text, consider allenai/scibert_scivocab_uncased or sentence-transformers/allenai-specter. Lower your threshold to 0.4 temporarily to diagnose whether the issue is threshold or embedding quality.

"The LLM-generated Cypher crashes with syntax errors."

Always log the generated Cypher before execution. Common issues:

• The LLM uses relationship types that don't exist in your schema
• The LLM uses backtick-quoted property names that Neo4j doesn't recognize
• The LLM generates MATCH (n:Entity {type: "DRUG"}) instead of MATCH (n:DRUG) when you've used APOC to add type labels

Feed the schema more precisely in your prompt, including example valid queries.

"The pipeline is too slow—extraction takes hours per document."

The bottleneck is almost certainly the LLM extraction call (one API call per chunk). Parallelize with asyncio and OpenAI's async client. For large corpora, batch chunks and use a cheaper model (gpt-4o-mini) with a more structured prompt. For extremely large corpora (millions of documents), consider a specialized NLP pipeline: spaCy with a domain-specific NER model for entity detection, and a dedicated relation extraction model, with the LLM reserved for ambiguous cases only.

"Graph traversal at 2 hops returns 10,000 nodes."

You have high-degree hub nodes. Implement the degree-filtered traversal from the scaling section. Also consider adding directionality constraints: MATCH (seed)-[:TREATS|:INTERACTS_WITH]->(hop1) instead of MATCH (seed)-[]->(hop1). Relationship type filtering is often the most effective degree control.

---

Summary & Next Steps

You've built a Graph RAG pipeline that does something standard RAG fundamentally cannot: it follows the connective tissue between pieces of information. Let's summarize the architecture:

1. Ingestion transforms raw text into a dual-representation: raw chunks in a vector index, and a typed knowledge graph where entities are nodes and named relationships are edges.

2. Query retrieval uses vector similarity to find entry points (seed nodes) in the graph, then expands outward via graph traversal to collect the neighborhood of relevant entities and relationships.

3. Context assembly combines the graph structure (serialized as relationship statements) with the source text chunks, giving the LLM both structured reasoning scaffolding and detailed evidence.

The key trade-offs to keep in mind:

• Extraction quality determines everything. A sparse or inconsistent graph is worse than no graph, because it creates false confidence. Invest heavily in your extraction prompt, your entity normalization, and your relationship ontology.
• Max hops is a risk dial. One hop is safe and fast. Two hops is usually the sweet spot for multi-hop queries. Three hops risks context explosion and latency spikes.
• Graph RAG doesn't replace vector RAG—it augments it. For simple factual lookups, vector search is faster and cheaper. Graph RAG pays off specifically when answers require tracing relationships across multiple entities.

Where to go next:

• Microsoft's GraphRAG library — Microsoft Research released an open-source GraphRAG implementation that uses community detection to build hierarchical summaries of graph communities. It's architecturally different from what we built (more summarization-focused, less traversal-focused) but worth studying.

• LangChain's Neo4j integration — langchain_community.graphs.neo4j_graph wraps many of these patterns with additional tooling. Useful once you understand the internals we've covered here.

• Temporal Graph RAG — Real-world knowledge graphs change over time. Drugs get withdrawn, trial phases advance, companies merge. Adding temporal edges and time-filtered traversal is the next sophistication layer.

• Graph neural networks for retrieval scoring — Instead of simple degree-filtered traversal, you can train a GNN to score node relevance given a query embedding. This is research-grade but increasingly accessible via PyTorch Geometric.

• Federated Graph RAG — Querying across multiple domain graphs (pharmaceutical × clinical × genomic) with entity alignment between them. Hard problem, fascinating architecture challenge.

The skills you've built here—entity extraction, knowledge graph construction, hybrid vector/graph retrieval—are foundational to almost every serious enterprise knowledge system being built right now. The companies that get this right will have AI systems that can actually reason over their internal knowledge, not just search it.