GraphRAG Patterns

Combine knowledge graphs with RAG for relationship-aware retrieval and reasoning.

When to Use

• Data has rich entity relationships

• Questions involve connections ("How is X related to Y?")

• Need multi-hop reasoning across documents

• Building over structured + unstructured data

• Want explainable retrieval paths

GraphRAG Architecture

┌──────────────────────────────────────────────────────────┐ │ Documents │ └─────────────────────────┬────────────────────────────────┘ │ ┌───────────────┼───────────────┐ │ │ │ ▼ ▼ ▼ ┌────────────┐ ┌────────────┐ ┌────────────┐ │ Entity │ │ Vector │ │ Text │ │ Extraction │ │ Embeddings │ │ Chunks │ └─────┬──────┘ └─────┬──────┘ └─────┬──────┘ │ │ │ ▼ │ │ ┌────────────┐ │ │ │ Knowledge │ │ │ │ Graph │ │ │ └─────┬──────┘ │ │ │ │ │ └───────────────┼───────────────┘ │ ▼ ┌─────────────────────┐ │ Hybrid Index │ │ (Graph + Vectors) │ └──────────┬──────────┘ │ ▼ ┌─────────────────────┐ │ Graph-Aware RAG │ └─────────────────────┘

Building the Knowledge Graph

Entity & Relationship Extraction

from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate

EXTRACTION_PROMPT = """Extract entities and relationships from the text.

Text: {text}

Return JSON: {{ "entities": [ {{"name": "...", "type": "PERSON|ORG|PRODUCT|CONCEPT|...", "description": "..."}} ], "relationships": [ {{"source": "...", "target": "...", "type": "WORKS_FOR|USES|RELATED_TO|...", "description": "..."}} ] }} """

def extract_graph_elements(text: str) -> dict: llm = ChatOpenAI(model="gpt-4", temperature=0) prompt = ChatPromptTemplate.from_template(EXTRACTION_PROMPT) chain = prompt | llm result = chain.invoke({"text": text}) return json.loads(result.content)

Store in Neo4j

from neo4j import GraphDatabase

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

def add_entity(self, entity: dict):
    with self.driver.session() as session:
        session.run("""
            MERGE (e:Entity {name: $name})
            SET e.type = $type, e.description = $description
            """,
            name=entity["name"],
            type=entity["type"],
            description=entity["description"]
        )

def add_relationship(self, rel: dict):
    with self.driver.session() as session:
        session.run("""
            MATCH (a:Entity {name: $source})
            MATCH (b:Entity {name: $target})
            MERGE (a)-[r:RELATES {type: $type}]->(b)
            SET r.description = $description
            """,
            source=rel["source"],
            target=rel["target"],
            type=rel["type"],
            description=rel["description"]
        )

def get_neighbors(self, entity: str, hops: int = 2) -> list:
    with self.driver.session() as session:
        result = session.run("""
            MATCH path = (e:Entity {name: $name})-[*1..$hops]-(related)
            RETURN path
            """,
            name=entity, hops=hops
        )
        return [record["path"] for record in result]

GraphRAG Retrieval Strategies

1. Entity-Centric Retrieval

def entity_centric_retrieve(query: str, graph: GraphStore, vectorstore) -> list: """Extract entities from query, expand via graph, retrieve chunks."""

# Extract entities from query
entities = extract_entities(query)

# Get graph neighbors
expanded_entities = set(entities)
for entity in entities:
    neighbors = graph.get_neighbors(entity, hops=2)
    expanded_entities.update(neighbors)

# Retrieve chunks mentioning these entities
chunks = []
for entity in expanded_entities:
    results = vectorstore.similarity_search(
        entity,
        k=3,
        filter={"entities": {"$contains": entity}}
    )
    chunks.extend(results)

return deduplicate(chunks)

2. Path-Based Retrieval

def path_retrieve(query: str, entity_a: str, entity_b: str, graph: GraphStore) -> str: """Find and explain paths between entities."""

with graph.driver.session() as session:
    result = session.run("""
        MATCH path = shortestPath(
            (a:Entity {name: $entity_a})-[*..5]-(b:Entity {name: $entity_b})
        )
        RETURN path, length(path) as hops
        ORDER BY hops
        LIMIT 5
        """,
        entity_a=entity_a, entity_b=entity_b
    )

    paths = []
    for record in result:
        path = record["path"]
        path_str = " -> ".join([node["name"] for node in path.nodes])
        paths.append(path_str)

return paths

3. Community-Based Retrieval (Microsoft GraphRAG)

from graspologic.partition import hierarchical_leiden

def build_communities(graph: GraphStore) -> dict: """Detect communities for hierarchical summarization."""

# Export graph to networkx
nx_graph = graph.to_networkx()

# Detect communities at multiple levels
communities = hierarchical_leiden(nx_graph, max_cluster_size=10)

# Summarize each community
community_summaries = {}
for community_id, members in communities.items():
    member_descriptions = [graph.get_entity(m)["description"] for m in members]
    summary = summarize_community(member_descriptions)
    community_summaries[community_id] = summary

return community_summaries

def community_retrieve(query: str, community_summaries: dict) -> list: """Search community summaries first, then drill down."""

# Find relevant communities
relevant = vectorstore.similarity_search(
    query,
    k=3,
    filter={"type": "community_summary"}
)

# Get entities from those communities
entities = []
for community in relevant:
    entities.extend(community.metadata["members"])

# Retrieve detailed chunks
return retrieve_by_entities(entities)

LangChain + Neo4j Integration

from langchain_community.graphs import Neo4jGraph from langchain.chains import GraphCypherQAChain

Connect to Neo4j

graph = Neo4jGraph( url="bolt://localhost:7687", username="neo4j", password="password" )

Natural language to Cypher

chain = GraphCypherQAChain.from_llm( llm=ChatOpenAI(model="gpt-4"), graph=graph, verbose=True, return_intermediate_steps=True )

Query in natural language

result = chain.invoke({ "query": "Who are the engineers working on Project Atlas?" })

Automatically generates: MATCH (p:Person)-[:WORKS_ON]->(proj:Project {name: 'Atlas'}) RETURN p

Hybrid Graph + Vector Pipeline

class GraphRAG: def init(self, graph: GraphStore, vectorstore, llm): self.graph = graph self.vectorstore = vectorstore self.llm = llm

def retrieve(self, query: str) -> list:
    # 1. Vector search for initial chunks
    vector_results = self.vectorstore.similarity_search(query, k=10)

    # 2. Extract entities from results
    entities = set()
    for doc in vector_results:
        entities.update(doc.metadata.get("entities", []))

    # 3. Expand via graph
    graph_context = []
    for entity in list(entities)[:5]:  # Limit expansion
        neighbors = self.graph.get_neighbors(entity, hops=1)
        for neighbor in neighbors:
            graph_context.append(f"{entity} -> {neighbor['relationship']} -> {neighbor['name']}")

    # 4. Combine contexts
    return {
        "chunks": vector_results,
        "graph_context": graph_context
    }

def generate(self, query: str, context: dict) -> str:
    prompt = f"""Answer based on the context.

    Text chunks:
    {self._format_chunks(context['chunks'])}

    Entity relationships:
    {chr(10).join(context['graph_context'])}

    Question: {query}
    """
    return self.llm.invoke(prompt).content

Best Practices

• Extract consistently - use same LLM/prompt for all documents

• Normalize entities - "AWS", "Amazon Web Services" → same node

• Limit graph depth - 2-3 hops usually sufficient

• Cache traversals - graph queries can be expensive

• Combine with vectors - graph alone misses semantic similarity

• Version your schema - entity/relationship types will evolve