GraphRAG in Python: Combining Knowledge Graphs and Vector Search for Multi-Hop Reasoning

Traditional Retrieval-Augmented Generation (RAG) relies on dense vector embeddings to match user queries with semantically similar document chunks. While vector search excels at direct factual lookups—such as retrieving a company’s refund policy or summarizing a specific technical specification—it consistently stumbles when faced with multi-hop reasoning across interconnected entities.

If an enterprise user asks, “Which vendor outages affected the infrastructure supporting Project Titan’s database cluster?”, standard vector similarity retrieves fragments mentioning “Project Titan”, “vendor outages”, or “database cluster”. However, it cannot connect the intermediate dots: Project Titan depends on Microservice A, which runs on Kubernetes Cluster B, which suffered an outage caused by Cloud Provider C.

This tutorial provides a complete engineering guide to implementing GraphRAG in Python. We will first build a zero-dependency in-memory GraphRAG pipeline using NetworkX and Sentence-Transformers, and then scale to an enterprise-grade production architecture using Neo4j and LangChain.

Architectural Overview: How GraphRAG Traversal Works

GraphRAG combines the semantic discovery of vector search with the structural rigor of a Knowledge Graph. Instead of passing only the raw vector top-k text chunks to the Large Language Model (LLM), GraphRAG executes a multi-stage retrieval flow:

Environment Setup & Dependencies

To implement both in-memory and database-backed GraphRAG pipelines, install the core dependencies in your virtual environment:

pip install networkx sentence-transformers langchain langchain-community langchain-neo4j neo4j pydantic python-dotenv

For the enterprise section, ensure you have a running Neo4j instance. You can launch a free local instance via Docker in seconds:

docker run \
    --name neo4j-graphrag \
    -p 7474:7474 -p 7687:7687 \
    -e NEO4J_AUTH=neo4j/SecretPassword123! \
    -e NEO4J_PLUGINS='["apoc"]' \
    -d neo4j:latest

Part 1: In-Memory GraphRAG with NetworkX and Sentence-Transformers

Before deploying a dedicated graph database, understanding the foundational mechanics of graph-augmented retrieval is critical. Here is a fully functional in-memory implementation that requires zero external infrastructure.

Building the Entity Graph & Vector Index

import numpy as np
import networkx as nx
from sentence_transformers import SentenceTransformer
from typing import List, Dict, Any, Tuple

class InMemoryGraphRAG:
    """Lightweight in-memory GraphRAG engine combining vector search and NetworkX."""

    def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
        self.encoder = SentenceTransformer(model_name)
        self.graph = nx.DiGraph()
        self.documents: List[str] = []
        self.embeddings: np.ndarray = np.empty((0, 384))

    def add_document(self, doc_text: str, triples: List[Tuple[str, str, str]]):
        """
        Adds a raw document chunk and its associated semantic triples.
        Triple format: (SourceEntity, Relationship, TargetEntity)
        """
        # 1. Store and embed document text
        doc_id = len(self.documents)
        self.documents.append(doc_text)
        vec = self.encoder.encode([doc_text])

        if self.embeddings.shape[0] == 0:
            self.embeddings = vec
        else:
            self.embeddings = np.vstack([self.embeddings, vec])

        # 2. Ingest entities and relationships into knowledge graph
        for src, rel, dst in triples:
            self.graph.add_node(src, type="Entity")
            self.graph.add_node(dst, type="Entity")
            self.graph.add_edge(src, dst, relation=rel, source_doc=doc_id)

    def retrieve(self, query: str, top_k_vectors: int = 2, max_hops: int = 1) -> Dict[str, Any]:
        """
        Executes hybrid retrieval:
        1. Semantic vector search across text chunks
        2. Graph traversal expanding seed entities discovered in the query
        """
        query_vec = self.encoder.encode([query])

        # Compute cosine similarity
        norm_docs = np.linalg.norm(self.embeddings, axis=1)
        norm_query = np.linalg.norm(query_vec)
        scores = np.dot(self.embeddings, query_vec.T).squeeze() / (norm_docs * norm_query + 1e-9)

        # Get top-k document indices
        top_indices = np.argsort(scores)[::-1][:top_k_vectors]
        matched_chunks = [{"doc_id": int(i), "score": float(scores[i]), "text": self.documents[i]} for i in top_indices]

        # Identify seed entities mentioned in query or matched documents
        seed_entities = set()
        for node in self.graph.nodes:
            if node.lower() in query.lower():
                seed_entities.add(node)

        # If no seeds in query directly, extract from top retrieved vector chunk
        if not seed_entities and matched_chunks:
            top_doc = matched_chunks[0]["text"]
            for node in self.graph.nodes:
                if node.lower() in top_doc.lower():
                    seed_entities.add(node)

        # Traverse graph up to max_hops
        expanded_triples = []
        for seed in seed_entities:
            # Outgoing relations
            for target in self.graph.successors(seed):
                edge_data = self.graph.get_edge_data(seed, target)
                expanded_triples.append({
                    "subject": seed,
                    "predicate": edge_data.get("relation", "RELATED_TO"),
                    "object": target
                })
                # 2-hop traversal if requested
                if max_hops > 1:
                    for second_target in self.graph.successors(target):
                        second_edge = self.graph.get_edge_data(target, second_target)
                        expanded_triples.append({
                            "subject": target,
                            "predicate": second_edge.get("relation", "RELATED_TO"),
                            "object": second_target
                        })

        return {
            "query": query,
            "vector_chunks": matched_chunks,
            "seed_entities": list(seed_entities),
            "graph_paths": expanded_triples
        }

# --- Demonstration & Testing ---
if __name__ == "__main__":
    rag = InMemoryGraphRAG()

    # Ingest document chunks with relational triples
    rag.add_document(
        "Project Titan is an enterprise payment processing engine led by engineer Marcus Vance.",
        [("Project Titan", "LED_BY", "Marcus Vance"), ("Project Titan", "USES", "Aurora PostgreDB")]
    )
    rag.add_document(
        "Aurora PostgreDB is hosted within AWS us-east-1 and managed by the CloudOps infrastructure squad.",
        [("Aurora PostgreDB", "HOSTED_IN", "AWS us-east-1"), ("Aurora PostgreDB", "MANAGED_BY", "CloudOps")]
    )
    rag.add_document(
        "AWS us-east-1 experienced a critical DNS degradation due to a fiber cut upstream from provider Lumen.",
        [("AWS us-east-1", "EXPERIENCED_OUTAGE", "DNS Degradation"), ("DNS Degradation", "CAUSED_BY", "Lumen Fiber Cut")]
    )

    # Query requiring 3-hop traversal
    query = "What external vendor issue affected Marcus Vance's project infrastructure?"
    result = rag.retrieve(query, top_k_vectors=2, max_hops=2)

    print("\n--- Seed Entities ---")
    print(result["seed_entities"])

    print("\n--- Discovered Graph Relational Paths ---")
    for triple in result["graph_paths"]:
        print(f"  ({triple['subject']}) --[{triple['predicate']}]--> ({triple['object']})")

Why This Outperforms Pure Vector Search

In standard vector search, querying "external vendor issue Marcus Vance project infrastructure" will score low on the chunk containing "Lumen Fiber Cut" because neither “Marcus Vance” nor “Project Titan” appears in that document.

With GraphRAG, the query locates Marcus Vance and Project Titan as seed entities, traverses through Aurora PostgreDB $\rightarrow$ AWS us-east-1 $\rightarrow$ DNS Degradation $\rightarrow$ Lumen Fiber Cut, and feeds this exact relational chain into the LLM synthesis prompt.

Part 2: Enterprise Knowledge Graph with Neo4j and LangChain

For production systems handling millions of documents, you need persistence, ACID transactions, indexed property graphs, and declarative graph querying via Cypher.

1. Connecting to Neo4j and Storing Graph Documents

import os
from langchain_community.graphs import Neo4jGraph
from langchain_core.documents import Document
from langchain_experimental.graph_transformers import LLMGraphTransformer
from langchain_openai import ChatOpenAI

# Initialize Neo4j graph connection
NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687")
NEO4J_USERNAME = os.getenv("NEO4J_USERNAME", "neo4j")
NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "SecretPassword123!")

graph = Neo4jGraph(
    url=NEO4J_URI,
    username=NEO4J_USERNAME,
    password=NEO4J_PASSWORD
)

# Automated Entity and Relationship Extraction via LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
graph_transformer = LLMGraphTransformer(
    llm=llm,
    allowed_nodes=["Project", "Person", "Database", "Region", "Incident", "Vendor"],
    allowed_relationships=["LED_BY", "USES", "HOSTED_IN", "MANAGED_BY", "EXPERIENCED", "CAUSED_BY"]
)

# Sample enterprise text corpus
raw_texts = [
    """
    Project Titan is an enterprise payment processing engine led by Marcus Vance. 
    It stores all critical settlement transactions in Aurora PostgreDB.
    """,
    """
    Aurora PostgreDB runs in cloud region AWS us-east-1 and is maintained by the CloudOps squad.
    Yesterday, AWS us-east-1 experienced a critical DNS incident.
    """,
    """
    The DNS incident in AWS us-east-1 was caused by a major fiber cut from Tier-1 network provider Lumen.
    """
]

docs = [Document(page_content=t.strip()) for t in raw_texts]

# Extract graph elements
graph_documents = graph_transformer.convert_to_graph_documents(docs)

# Add elements into Neo4j
graph.add_graph_documents(graph_documents, include_source=True)
print(f"Ingested {len(graph_documents)} documents into Neo4j Knowledge Graph.")

2. Multi-Hop Question Answering with Cypher Generation

Once your knowledge graph is populated, use GraphCypherQAChain to allow the LLM to inspect the schema, translate natural language into Cypher queries, execute them against Neo4j, and return grounded answers:

from langchain_community.chains.graph_qa.cypher import GraphCypherQAChain
from langchain_openai import ChatOpenAI

# Refresh schema dynamically from database
graph.refresh_schema()

cypher_chain = GraphCypherQAChain.from_llm(
    cypher_llm=ChatOpenAI(model="gpt-4o", temperature=0),
    qa_llm=ChatOpenAI(model="gpt-4o", temperature=0),
    graph=graph,
    verbose=True,
    return_intermediate_steps=True
)

query = "Which vendor caused the incident that impacted the database used by Marcus Vance's project?"
response = cypher_chain.invoke({"query": query})

print("\n--- Generated Cypher Query ---")
print(response["intermediate_steps"][0]["query"])

print("\n--- Database Context Returned ---")
print(response["intermediate_steps"][1]["context"])

print("\n--- Final Synthesized Response ---")
print(response["result"])

When executed, the LLM generates a multi-hop traversal query similar to:

MATCH (p:Person {id: 'Marcus Vance'})<-[:LED_BY]-(proj:Project)-[:USES]->(db:Database)
MATCH (db)-[:HOSTED_IN]->(r:Region)-[:EXPERIENCED]->(inc:Incident)-[:CAUSED_BY]->(v:Vendor)
RETURN v.id AS Vendor, inc.id AS Incident, db.id AS Database

Part 3: Hybrid Retrieval Architecture (Vector + Cypher)

In enterprise production environments, the most resilient pattern is Hybrid VectorCypher Retrieval.

1. Dense Vector Search finds the most relevant starting nodes or passage chunks based on fuzzy semantic similarity.
2. Graph Expansion follows relationships 1 to 2 hops away to retrieve context that vector search alone would never locate.

from langchain_community.vectorstores import Neo4jVector
from langchain_openai import OpenAIEmbeddings

# Build vector index inside Neo4j alongside graph nodes
vector_index = Neo4jVector.from_existing_graph(
    embedding=OpenAIEmbeddings(),
    url=NEO4J_URI,
    username=NEO4J_USERNAME,
    password=NEO4J_PASSWORD,
    index_name="entity_vector_index",
    node_label="Document",
    text_node_properties=["text"],
    embedding_node_>

Latency, Complexity, and Cost Trade-offs

Deploying GraphRAG introduces unique architectural trade-offs compared to traditional vector storage:

Production Best Practices for GraphRAG

1. Constrain Entity Schemas: Always provide allowed_nodes and allowed_relationships to LLMGraphTransformer. Unconstrained extraction causes entity explosion (e.g., generating 40 variations of "Database").
2. Implement Entity Resolution: Map synonyms ("AWS us-east-1", "us-east-1", "N. Virginia Region") to a single canonical node ID during ingestion.
3. Limit Traversal Hops: Bound Cypher queries to a maximum of 2 hops (MATCH (s)-[*1..2]->(t)). Unbounded path queries (MATCH (s)-[*]->(t)) cause memory spikes and severe graph engine lockups.
4. Cache Cypher Queries: Use deterministic parameter hashing to cache Cypher queries for recurring query structures.

Frequently Asked Questions