Graph RAG

When to Use Graph RAG vs Vector RAG

Rule of thumb: use graph RAG when the answer requires connecting pieces of information across documents; use vector RAG when the answer lives inside a single chunk.

Entity and Triple Extraction

# Using Claude to extract (subject, predicate, object) triples
import anthropic, json, re

client = anthropic.Anthropic()

EXTRACT_PROMPT = """Extract all factual triples from the text as JSON.
Schema: [{"s": "...", "p": "...", "o": "...", "s_type": "Person|Org|Product|Concept", "o_type": "..."}]
Only extract facts explicitly stated. Normalize entity names (no pronouns).

Text:
{text}

Return JSON only."""

def extract_triples(text: str) -> list[dict]:
    msg = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=2000,
        messages=[{"role": "user", "content": EXTRACT_PROMPT.format(text=text)}],
    )
    raw = msg.content[0].text
    return json.loads(re.search(r"\[.*\]", raw, re.DOTALL).group(0))

For high-volume extraction, use a cheaper model (claude-haiku-4-5 or gpt-4o-mini) and batch 10-20 chunks per call. Cache by sha256(chunk) to avoid re-extraction.

Neo4j + Embeddings Hybrid Store

from neo4j import GraphDatabase
from openai import OpenAI

driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "pass"))
oai = OpenAI()

CREATE_SCHEMA = """
CREATE CONSTRAINT entity_name IF NOT EXISTS FOR (e:Entity) REQUIRE e.name IS UNIQUE;
CREATE VECTOR INDEX entity_embed IF NOT EXISTS
  FOR (e:Entity) ON (e.embedding)
  OPTIONS {indexConfig: {`vector.dimensions`: 1536, `vector.similarity_function`: 'cosine'}};
"""

def embed(text: str) -> list[float]:
    return oai.embeddings.create(model="text-embedding-3-small", input=text).data[0].embedding

def upsert_triple(tx, s, p, o, s_type, o_type, source_chunk_id):
    tx.run("""
        MERGE (a:Entity {name: $s}) SET a.type = $s_type, a.embedding = $s_emb
        MERGE (b:Entity {name: $o}) SET b.type = $o_type, b.embedding = $o_emb
        MERGE (a)-[r:REL {type: $p}]->(b)
        SET r.source = $src
    """, s=s, o=o, p=p, s_type=s_type, o_type=o_type,
        s_emb=embed(s), o_emb=embed(o), src=source_chunk_id)

with driver.session() as s:
    s.run(CREATE_SCHEMA)
    for t in triples:
        s.execute_write(upsert_triple, t["s"], t["p"], t["o"],
                        t["s_type"], t["o_type"], chunk_id)

Graph-augmented retrieval (entity seed + n-hop expansion)

// Find entities semantically similar to the query, then expand 2 hops
CALL db.index.vector.queryNodes('entity_embed', 5, $query_embedding)
YIELD node AS seed, score
MATCH path = (seed)-[*1..2]-(neighbor:Entity)
WITH seed, neighbor, score, [r IN relationships(path) | r.type] AS rels
RETURN seed.name, neighbor.name, rels, score
ORDER BY score DESC LIMIT 30;

Microsoft GraphRAG (Community Summarization)

GraphRAG's insight: after building the entity graph, run hierarchical community detection (Leiden), summarize each community with an LLM, and use those summaries for global queries.

# pip install graphrag
# graphrag init --root ./ragtest
# Configure settings.yaml with OpenAI/Azure keys, then:
#   graphrag index --root ./ragtest
#
# Query modes:
#   graphrag query --method local  "What did Alice work on?"
#   graphrag query --method global "Summarize all regulatory themes in the corpus"
#   graphrag query --method drift  "How do safety issues relate to recalls?"

from graphrag.query.structured_search.local_search.search import LocalSearch
from graphrag.query.structured_search.global_search.search import GlobalSearch

# Local = entity-centric (start from entities matching the query, walk graph)
# Global = map-reduce over community summaries (thematic / corpus-wide)
# DRIFT  = hybrid: local seed, global expansion for connective tissue

When to pick which mode:

• Local: "What's X's role?", "Who did X work with?" - starts from named entities.
• Global: "Main themes", "Recurring risks", "Summarize the corpus" - needs community reports.
• DRIFT: multi-hop exploratory questions where local and global both matter.

HippoRAG (Personalized PageRank)

HippoRAG builds a passage-to-entity graph once, then at query time:

1. NER on query to find seed entities.
2. Run Personalized PageRank (PPR) with seeds as the teleport distribution.
3. Rank passages by aggregated PPR mass of their entities.

# pip install hipporag
from hipporag import HippoRAG

hipporag = HippoRAG(
    save_dir="./hippo_index",
    llm_model_name="gpt-4o-mini",
    embedding_model_name="nvidia/NV-Embed-v2",
)

hipporag.index(docs=["passage 1...", "passage 2...", "passage 3..."])
answers = hipporag.rag_qa(queries=["Which researcher's student works on X?"])

Use HippoRAG when you need single-step multi-hop retrieval without building explicit Cypher queries.

LlamaIndex Property Graph Index

from llama_index.core import PropertyGraphIndex, Document
from llama_index.core.indices.property_graph import (
    SchemaLLMPathExtractor, ImplicitPathExtractor,
)
from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore
from llama_index.llms.anthropic import Anthropic
from llama_index.embeddings.openai import OpenAIEmbedding
from typing import Literal

Entities = Literal["Person", "Org", "Product", "Concept"]
Relations = Literal["WORKS_AT", "FOUNDED", "USES", "RELATED_TO"]

graph_store = Neo4jPropertyGraphStore(
    username="neo4j", password="pass", url="bolt://localhost:7687",
)

index = PropertyGraphIndex.from_documents(
    [Document(text=t) for t in corpus],
    property_graph_store=graph_store,
    kg_extractors=[
        SchemaLLMPathExtractor(
            llm=Anthropic(model="claude-opus-4-5"),
            possible_entities=Entities,
            possible_relations=Relations,
            strict=True,  # drop triples not matching schema
        ),
        ImplicitPathExtractor(),  # adds MENTIONS edges from chunks to entities
    ],
    embed_model=OpenAIEmbedding(model="text-embedding-3-small"),
    show_progress=True,
)

# Query with graph + vector hybrid
retriever = index.as_retriever(
    include_text=True,
    similarity_top_k=10,
    path_depth=2,   # expand 2 hops from seed entities
)
nodes = retriever.retrieve("Which founders worked at companies that use Kafka?")

LangChain Graph Retriever

from langchain_neo4j import Neo4jGraph, GraphCypherQAChain
from langchain_anthropic import ChatAnthropic

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

chain = GraphCypherQAChain.from_llm(
    llm=ChatAnthropic(model="claude-opus-4-5"),
    graph=graph,
    verbose=True,
    allow_dangerous_requests=True,  # LLM writes Cypher; restrict DB user to read-only
    validate_cypher=True,
    return_intermediate_steps=True,
)

result = chain.invoke({"query": "Who reports to Alice's manager?"})
# Inspect result["intermediate_steps"][0]["query"] for the generated Cypher

Safety: always give the Neo4j role only MATCH/read privileges; never let an LLM write or delete.

Hybrid Graph + Vector Pipeline

def hybrid_retrieve(query: str, k_vec: int = 10, k_graph: int = 10):
    # 1. Semantic top-K from chunk vector index
    vec_hits = vectorstore.similarity_search(query, k=k_vec)

    # 2. Graph traversal seeded by query entities
    entities = extract_entities(query)
    with driver.session() as s:
        graph_hits = s.run("""
            MATCH (e:Entity) WHERE e.name IN $ents
            MATCH (e)-[*1..2]-(n:Entity)<-[:MENTIONS]-(c:Chunk)
            RETURN DISTINCT c.id AS id, c.text AS text
            LIMIT $k
        """, ents=entities, k=k_graph).data()

    # 3. Rerank the union with Cohere or a cross-encoder
    return rerank(query, vec_hits + graph_hits, top_n=6)

Anti-Patterns

Production Checklist

• Schema for entities and relations documented and enforced
• Entity canonicalization step (dedup aliases before insert)
• Provenance (source chunk + offsets) stored on every edge
• Read-only DB credentials for LLM-generated Cypher
• Incremental graph updates on doc ingest (no full rebuilds in prod)
• Hybrid retriever (graph + vector + rerank) benchmarked vs vector-only
• Community summaries refreshed on a schedule (GraphRAG)
• Latency budget: graph traversal capped (e.g., LIMIT 50, depth <= 2)