Neo4j Graphrag Skill

Neo4j GraphRAG Skill

When to Use

• Building GraphRAG retrieval pipelines with neo4j-graphrag Python package
• Choosing between VectorRetriever, HybridRetriever, VectorCypherRetriever, HybridCypherRetriever
• Writing retrieval_query Cypher fragments for graph-augmented context
• Wiring retriever + LLM into a GraphRAG pipeline
• Using LLM-routed multi-retriever with ToolsRetriever
• Debugging low retrieval quality
• Integrating Neo4j with LangChain, LlamaIndex, or Haystack

When NOT to Use

• KG construction from documents → neo4j-document-import-skill
• Plain vector/semantic search without graph traversal → neo4j-vector-index-skill
• Hybrid search that combines vector with fulltext or other ranked sources → neo4j-vector-index-skill
• GDS algorithms (PageRank, Louvain, node embeddings) → neo4j-gds-skill
• Agent long-term memory → neo4j-agent-memory-skill
• Writing raw Cypher queries → neo4j-cypher-skill

---

Retriever Selection

Has fulltext index?
  YES → Hybrid variants (HybridRetriever / HybridCypherRetriever)
  NO  → Vector variants (VectorRetriever / VectorCypherRetriever)

Need graph traversal after vector lookup?
  YES → Cypher variants (VectorCypherRetriever / HybridCypherRetriever)
  NO  → plain variants

Natural-language-to-Cypher?        → Text2CypherRetriever (no embedder needed)
LLM should route between retrievers? → ToolsRetriever
Vectors stored in external DB?      → WeaviateNeo4jRetriever / PineconeNeo4jRetriever / QdrantNeo4jRetriever

---

Install

pip install neo4j-graphrag[openai]        # OpenAI LLM + embeddings
pip install neo4j-graphrag[anthropic]     # Anthropic Claude
pip install neo4j-graphrag[google]        # Vertex AI / Gemini
pip install neo4j-graphrag[bedrock]       # Amazon Bedrock (boto3)
pip install neo4j-graphrag[cohere]        # Cohere
pip install neo4j-graphrag[mistralai]     # MistralAI
pip install neo4j-graphrag[ollama]        # Ollama (local)
pip install neo4j-graphrag[weaviate]      # Weaviate external retriever
pip install neo4j-graphrag[pinecone]      # Pinecone external retriever
pip install neo4j-graphrag[qdrant]        # Qdrant external retriever

Requires: Python >= 3.10, neo4j >= 5.17.0 (driver 6.x supported).

---

Step 2 — Choose Retriever

Has fulltext index? YES → Hybrid variants (better recall)
                   NO  → Vector variants (baseline)

Needs graph context after vector lookup? YES → Cypher variants
                                         NO  → plain variants

For natural-language-to-Cypher? → Text2CypherRetriever (no embedder needed)
For multi-tool LLM routing?     → ToolsRetriever
Using external vector DB?       → WeaviateNeo4jRetriever / PineconeNeo4jRetriever / QdrantNeo4jRetriever

For custom Cypher hybrid search outside the neo4j-graphrag retriever APIs, use neo4j-vector-index-skill.

Vector backend selection [v1.16+, auto]: on Neo4j 2026.01+ all four vector/hybrid retrievers auto-route through the Cypher 25 SEARCH ... WHERE clause when filters are SEARCH-compatible (simple AND comparisons) and all filter props are declared in the index WITH [n.prop] list. $or, $in, $like, or undeclared props → automatic fallback to db.index.vector.queryNodes() procedure path (with warning log). Declare filterable properties via filterable_properties=[...] on create_vector_index().

---

Step 3 — Create Indexes (run once)

// Vector index (all retrievers need this)
CREATE VECTOR INDEX chunk_embedding IF NOT EXISTS
FOR (c:Chunk) ON (c.embedding)
OPTIONS { indexConfig: {
  `vector.dimensions`: 1536,
  `vector.similarity_function`: 'cosine'
} };

// Fulltext index (Hybrid retrievers only)
CREATE FULLTEXT INDEX chunk_fulltext IF NOT EXISTS
FOR (c:Chunk) ON EACH [c.text];

// Confirm ONLINE before ingesting:
SHOW INDEXES YIELD name, state
WHERE name IN ['chunk_embedding', 'chunk_fulltext']
RETURN name, state;
// Both must show state = 'ONLINE'

If index not ONLINE: wait, poll every 5s. Do NOT start ingestion until ONLINE.

---

Step 4 — Core Pattern (HybridCypherRetriever)

from neo4j import GraphDatabase
from neo4j_graphrag.embeddings import OpenAIEmbeddings
from neo4j_graphrag.generation import GraphRAG
from neo4j_graphrag.llm import OpenAILLM
from neo4j_graphrag.retrievers import HybridCypherRetriever

driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USERNAME, NEO4J_PASSWORD))
embedder = OpenAIEmbeddings(model="text-embedding-3-large")  # OPENAI_API_KEY from env

# retrieval_query: Cypher fragment executed after the vector/fulltext lookup.
# Auto-injected variables:  node  (matched node)   score  (similarity float)
# MUST include a RETURN clause.  score must appear in RETURN.
retrieval_query = """
MATCH (node)<-[:HAS_CHUNK]-(article:Article)
OPTIONAL MATCH (article)-[:MENTIONS]->(org:Organization)
RETURN node.text AS chunk_text,
       article.title AS article_title,
       collect(DISTINCT org.name) AS mentioned_organizations,
       score
"""

retriever = HybridCypherRetriever(
    driver=driver,
    vector_index_name="chunk_embedding",
    fulltext_index_name="chunk_fulltext",
    retrieval_query=retrieval_query,
    embedder=embedder,
)

llm = OpenAILLM(model_name="gpt-4.1", model_params={"temperature": 0})

rag = GraphRAG(
    retriever=retriever,
    llm=llm,
)

response = rag.search(
    query_text="Who does Alice work for?",
    retriever_config={"top_k": 5},
)
print(response.answer)
driver.close()

---

VectorCypherRetriever

from neo4j_graphrag.retrievers import VectorCypherRetriever

retriever = VectorCypherRetriever(
    driver=driver,
    index_name="chunk_embedding",
    retrieval_query=retrieval_query,
    embedder=embedder,
)

response = rag.search(
    query_text="What happened at Apple?",
    retriever_config={"top_k": 10},
)

---

Text2CypherRetriever

Translates natural language to Cypher using an LLM. No embedder required.

Security (v1.16.0+): Every LLM-generated Cypher is run through EXPLAIN first.

Any statement classified as write/destructive raises Text2CypherRetrievalError instead

of executing — prevents prompt-injection attacks.

from neo4j_graphrag.retrievers import Text2CypherRetriever

retriever = Text2CypherRetriever(
    driver=driver,
    llm=OpenAILLM(model_name="gpt-4.1"),
    neo4j_schema=None,       # None = auto-fetch schema from DB; pass string to trim
    examples=[
        "Q: Who works at Neo4j? A: MATCH (p:Person)-[:WORKS_AT]->(c:Company {name:'Neo4j'}) RETURN p.name"
    ],
)
results = retriever.search(query_text="Which people work at Neo4j?")

---

ToolsRetriever (LLM-routed multi-retriever)

from neo4j_graphrag.retrievers import ToolsRetriever

tools_retriever = ToolsRetriever(
    llm=llm,
    retrievers=[vector_retriever, text2cypher_retriever],
)
# LLM decides which retriever(s) to invoke per query

# Convert any retriever to a standalone Tool:
tool = vector_retriever.convert_to_tool()

---

Filters (pre-filter before vector search)

results = retriever.search(
    query_text="quarterly earnings",
    top_k=5,
    filters={
        "date": {"$gte": "2024-01-01"},
        "source": {"$eq": "10-K"},
    },
)
# Operators: $eq  $ne  $lt  $lte  $gt  $gte  $between  $in  $like  $ilike

---

query_params (parameterized retrieval_query)

retrieval_query = """
MATCH (node)<-[:HAS_CHUNK]-(a:Article)-[:MENTIONS]->(org:Organization {name: $entity_name})
RETURN node.text, a.title, score
"""

# Pass via retriever.search directly:
results = retriever.search(
    query_text="What happened at Apple?",
    top_k=10,
    query_params={"entity_name": "Apple"},
)

# Or via GraphRAG.search:
response = rag.search(
    query_text="What happened at Apple?",
    retriever_config={"top_k": 10, "query_params": {"entity_name": "Apple"}},
)

---

Cypher 25 SEARCH Clause (v1.16.0, Neo4j 2026.x+)

# Enable SEARCH clause syntax in vector/hybrid retrievers (requires Neo4j 2026+)
retriever = VectorRetriever(
    driver=driver,
    index_name="chunk_embedding",
    embedder=embedder,
    use_search_clause=True,
)

---

ORDER BY on Cypher Retrievers (v1.16.0)

results = retriever.search(
    query_text="...",
    top_k=10,
    order_by="score DESC",
)

If neo4j_schema=None: retriever fetches schema automatically. For large schemas, pass a trimmed string to reduce LLM prompt size.

Destructive-query guard [v1.16+]: Text2CypherRetriever runs EXPLAIN on the generated Cypher before execution and rejects queries that produce writes (CREATE, MERGE, DELETE, SET, REMOVE, etc.). LLM-generated writes are never executed against the graph.

---

Custom Prompt Template

from neo4j_graphrag.generation.prompts import RagTemplate

template = RagTemplate(
    template="""Answer using ONLY the context below.
Context: {context}
Question: {query_text}
Answer:""",
    expected_inputs=["context", "query_text"],
)

rag = GraphRAG(retriever=retriever, llm=llm, prompt_template=template)

---

return_context and response_fallback

response = rag.search(
    query_text="...",
    retriever_config={"top_k": 5},
    return_context=True,                        # include raw retrieved chunks
    response_fallback="No relevant context.",   # skip LLM call if retriever returns nothing
)
print(response.answer)
print(response.retriever_result)    # RawSearchResult when return_context=True

---

Message History (multi-turn)

from neo4j_graphrag.message_history import InMemoryMessageHistory

history = InMemoryMessageHistory()
r1 = rag.search(query_text="Who is Alice?", message_history=history)
r2 = rag.search(query_text="Where does she work?", message_history=history)

---

External Retrievers

# --- Weaviate ---
from neo4j_graphrag.retrievers import WeaviateNeo4jRetriever
import weaviate

weaviate_client = weaviate.connect_to_local()
retriever = WeaviateNeo4jRetriever(
    driver=driver,
    client=weaviate_client,
    collection="Chunk",
    id_property_external="neo4j_id",
    id_property_neo4j="id",
    retrieval_query=retrieval_query,
    node_label_neo4j="Chunk",       # optional: speeds up Neo4j lookup
)

# --- Pinecone ---
from neo4j_graphrag.retrievers import PineconeNeo4jRetriever
from pinecone import Pinecone

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
retriever = PineconeNeo4jRetriever(
    driver=driver,
    client=pc,
    index_name="my-index",
    id_property_neo4j="id",
    retrieval_query=retrieval_query,
)

# --- Qdrant ---
from neo4j_graphrag.retrievers import QdrantNeo4jRetriever
from qdrant_client import QdrantClient

retriever = QdrantNeo4jRetriever(
    driver=driver,
    client=QdrantClient(url="http://localhost:6333"),
    collection_name="Chunk",
    id_property_external="neo4j_id",
    id_property_neo4j="id",
    id_property_getter=lambda hit: hit.payload["neo4j_id"],  # custom ID extraction
    retrieval_query=retrieval_query,
)

---

LLM Providers

All implement LLMBase. All support sync + async, tool calling, and automatic rate limiting.

from neo4j_graphrag.llm import (
    OpenAILLM, AzureOpenAILLM, AnthropicLLM, VertexAILLM,
    MistralAILLM, CohereLLM, OllamaLLM, BedrockLLM,
)

llm = OpenAILLM(model_name="gpt-4.1", model_params={"temperature": 0})
llm = AnthropicLLM(model_name="claude-3-5-sonnet-20241022")
llm = VertexAILLM(model_name="gemini-2.0-flash")
llm = OllamaLLM(model_name="llama3")           # no API key needed
llm = BedrockLLM(model_id="anthropic.claude-3-5-sonnet-20241022-v2:0")

# Token usage tracking (v1.15.0+)
response = llm.invoke("Hello")
# response.usage → LLMUsage(request_tokens=N, response_tokens=M, total_tokens=T)

# Graceful resource cleanup (v1.16.0+)
llm.close()         # sync
await llm.aclose()  # async

---

Embedder Providers

All include automatic rate limiting with tenacity exponential backoff.

from neo4j_graphrag.embeddings import (
    OpenAIEmbeddings, VertexAIEmbeddings, CohereEmbeddings,
    OllamaEmbeddings, SentenceTransformerEmbeddings, BedrockEmbeddings,
)

embedder = OpenAIEmbeddings(model="text-embedding-3-large")   # 3072 dims
embedder = OpenAIEmbeddings(model="text-embedding-3-small")   # 1536 dims
embedder = SentenceTransformerEmbeddings(model="all-MiniLM-L6-v2")  # 384 dims, local
embedder = BedrockEmbeddings(model_id="amazon.titan-embed-text-v2:0")

---

Index Setup

from neo4j_graphrag.indexes import create_vector_index

# Vector index — adjust dimensions to match your embedding model
create_vector_index(
    driver,
    name="chunk_embedding",
    label="Chunk",
    embedding_property="embedding",
    dimensions=1536,
    similarity_fn="cosine",       # or "euclidean"
)

# Fulltext index (run as Cypher)
# CREATE FULLTEXT INDEX chunk_fulltext IF NOT EXISTS
#   FOR (c:Chunk) ON EACH [c.text]

---

Schema Inspection

from neo4j_graphrag.schema import get_schema, get_structured_schema

schema_str = get_schema(driver, sample=1000)           # human-readable string
schema_dict = get_structured_schema(driver, sample=1000)  # dict with labels/rels/props

---

Common Errors

---

Verification Checklist

• [ ] neo4j-graphrag (not neo4j-genai) installed; neo4j >= 5.17.0 driver
• [ ] Vector index ONLINE before ingesting embeddings or running retriever
• [ ] Fulltext index ONLINE if using Hybrid variants
• [ ] Embedding dims in create_vector_index match the embedder output
• [ ] retrieval_query returns node and score in RETURN (not re-declared)
• [ ] query_params passed via retriever_config on rag.search() (not on retriever constructor)
• [ ] API keys in env vars; never hardcoded
• [ ] llm.close() called when done to release resources

---

References

Load on demand:

• neo4j-graphrag package docs
• RAG & GraphRAG user guide
• KG Builder user guide
• GitHub — neo4j-graphrag-python
• Examples folder

neo4j-graphrag-skill

Skill for building GraphRAG retrieval pipelines on Neo4j using the neo4j-graphrag Python package (formerly neo4j-genai).

Covers:

• Retriever selection: VectorRetriever, HybridRetriever, VectorCypherRetriever, HybridCypherRetriever, Text2CypherRetriever
• retrieval_query — Cypher fragments for post-vector graph traversal; node and score auto-injection
• query_params — parameterized retrieval queries
• Pipeline wiring: GraphRAG(retriever=..., llm=...) with .search()
• Embedder setup: OpenAIEmbeddings, VertexAIEmbeddings, etc.
• Index prerequisites: vector index + fulltext index for Hybrid retrievers
• Custom prompt templates via prompt_template
• LangChain (langchain-neo4j), LlamaIndex, and Haystack integrations
• Pre-filter vector search with filters=
• Text2CypherRetriever for exact structured queries (counts, lookups)

Version / compatibility:

• neo4j-graphrag v1.7+ (renamed from neo4j-genai; uninstall old package if present)
• Python ≥ 3.10

Not covered:

• KG construction from documents (SimpleKGPipeline) → neo4j-document-import-skill
• Pure-Cypher GraphRAG with ai.text.* → neo4j-genai-plugin-skill
• Vector index creation → neo4j-vector-index-skill

Install:

pip install neo4j-graphrag

npx skills add https://github.com/neo4j-contrib/neo4j-skills --skill neo4j-graphrag-skill

Or paste this link into your coding assistant: https://github.com/neo4j-contrib/neo4j-skills/tree/main/neo4j-graphrag-skill

SimpleKGPipeline Reference

Full Constructor

from neo4j_graphrag.experimental.pipeline.kg_builder import SimpleKGPipeline

pipeline = SimpleKGPipeline(
    llm,                          # LLMInterface — used for entity/rel extraction
    driver,                       # Neo4j driver
    embedder,                     # Embedder — for chunk embeddings
    from_file: bool = True,       # True: pass file_path; False: pass text=
    entities: list = None,        # simple: ["Person", "Org"] or detailed dicts
    relations: list = None,       # simple: ["WORKS_AT"] or detailed dicts
    schema: dict = None,          # explicit schema (overrides entities/relations)
    perform_entity_resolution: bool = True,
    neo4j_database: str = None,
    on_error: str = "IGNORE",     # "RAISE" or "IGNORE"
    prompt_template: str = None,  # custom extraction prompt
    chunk_size: int = 1000,
    chunk_overlap: int = 200,
)

Schema Modes

Simple string lists (auto-generates extraction prompt):

entities=["Person", "Organization", "Location"]
relations=["WORKS_AT", "LOCATED_IN", "KNOWS"]

Detailed dicts (adds LLM guidance per type):

entities=[
    {"label": "Person", "description": "A human individual", "properties": [{"name": "name", "type": "str"}]},
    {"label": "Organization", "description": "A company or institution"},
]
relations=[
    {"label": "WORKS_AT", "description": "Person employed by Organization",
     "properties": [{"name": "since", "type": "str"}]},
]

Schema patterns (restrict which entity pairs a relation connects):

schema={
    "node_types": [...],
    "relationship_types": [...],
    "patterns": [
        {"start": "Person", "end": "Organization", "relation": "WORKS_AT"},
    ]
}

Schema modes (set via schema param):

• EXTRACTED (default) — LLM infers schema from text
• FREE — unguided extraction; no schema constraint
• Custom — pass explicit schema dict

Chunking Options

from neo4j_graphrag.experimental.components.text_splitters.fixed_size_splitter import FixedSizeSplitter

pipeline = SimpleKGPipeline(
    ...,
    chunk_size=800,      # tokens per chunk
    chunk_overlap=150,   # overlap between consecutive chunks
)
# For LangChain splitters:
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Pass custom splitter via pipeline component wiring (advanced — see pipeline docs)

Graph Structure Created

(Document {source, metadata})
  -[:FROM_DOCUMENT]->
(Chunk {text, embedding, index, ...})
  -[:HAS_ENTITY]->
(Entity {name, label, description, ...})
  -[:<RELATION_TYPE>]->
(Entity)

entity_resolution

When perform_entity_resolution=True (default): similar entity nodes are merged using LLM judgment. Increases accuracy, slower. Set False for initial development, True for production ingestion.

Structured Output

Use with OpenAI or VertexAI to enforce JSON schema at API level (reduces parsing errors):

llm = OpenAILLM(model_name="gpt-4o", model_params={"use_structured_output": True})

Not available for Anthropic, Ollama, Mistral (as of v1.x).

Batch Processing Multiple Documents

import asyncio

async def ingest_all(docs):
    for doc in docs:
        await pipeline.run_async(
            text=doc["text"],
            document_metadata={"title": doc["title"], "date": doc["date"]},
        )

asyncio.run(ingest_all(documents))

Sequential by default. For parallel: use asyncio.gather() — monitor Neo4j connection pool.

Debug: Inspect LLM Extraction

# Force errors to surface:
pipeline = SimpleKGPipeline(..., on_error="RAISE")

# Check what was extracted:
result = asyncio.run(pipeline.run_async(text="Alice works at Neo4j in London."))
# Inspect result for extracted entities and relations

Knowledge Graph Construction — Advanced Reference

Supplements SKILL.md Step 11. Advanced reference — use when SimpleKGPipeline defaults are insufficient.

---

Text Splitting Strategies

FixedSizeSplitter

from neo4j_graphrag.experimental.components.text_splitters.fixed_size_splitter import FixedSizeSplitter

splitter = FixedSizeSplitter(chunk_size=500, chunk_overlap=100)

chunk_size = max chars per chunk. chunk_overlap = chars shared between consecutive chunks — preserves context across boundaries.

Custom TextSplitter (section-aware)

from neo4j_graphrag.experimental.components.text_splitters.base import TextSplitter, TextChunk, TextChunks

class SectionSplitter(TextSplitter):
    def run(self, text: str) -> TextChunks:
        sections = re.split(r"^== ", text, flags=re.MULTILINE)
        chunks = [TextChunk(uid=i, text=s.strip()) for i, s in enumerate(sections) if s.strip()]
        return TextChunks(chunks=chunks)

pipeline = SimpleKGPipeline(..., text_splitter=SectionSplitter())

LangChain Adapter

from neo4j_graphrag.experimental.components.text_splitters.langchain import LangChainTextSplitterAdapter
from langchain_text_splitters import CharacterTextSplitter

lc_splitter = CharacterTextSplitter(separator="\n\n", chunk_size=500, chunk_overlap=100)
splitter = LangChainTextSplitterAdapter(lc_splitter)
pipeline = SimpleKGPipeline(..., text_splitter=splitter)

Install: pip install langchain-text-splitters

---

LexicalGraphConfig — Rename Graph Labels

Default graph model: Document -[:FROM_DOCUMENT]-> Chunk -[:FROM_CHUNK]-> __Entity__

Override any label or relationship name:

from neo4j_graphrag.experimental.components.kg_writer import LexicalGraphConfig

config = LexicalGraphConfig(
    id_prefix="lesson",                              # prefix for generated IDs
    document_node_label="Lesson",                    # default: "Document"
    chunk_node_label="Section",                      # default: "Chunk"
    chunk_to_document_relationship_type="PART_OF",   # default: "FROM_DOCUMENT"
    next_chunk_relationship_type="NEXT_SECTION",     # default: "NEXT_CHUNK"
    node_to_chunk_relationship_type="FROM_SECTION",  # default: "FROM_CHUNK"
    chunk_embedding_property="vector",               # default: "embedding"
)
pipeline = SimpleKGPipeline(..., lexical_graph_config=config)

Inspect constructed graph:

MATCH (d:Document)<-[:FROM_DOCUMENT]-(c:Chunk)
RETURN d.path, c.index, c.text, size(c.text)
ORDER BY d.path, c.index

---

Entity Resolution — Strategies

Default (inline, identical name merge)

perform_entity_resolution=True (default): merges nodes sharing same label + identical name during ingestion. Fast, exact-match only.

Disable (keep all duplicates)

pipeline = SimpleKGPipeline(..., perform_entity_resolution=False)

Use during development for speed. Risk: duplicate entity nodes.

Post-Processing Resolvers (run after ingestion)

FuzzyMatchResolver

Merges entities with same label + similar name using RapidFuzz edit distance:

from neo4j_graphrag.experimental.components.resolver import FuzzyMatchResolver

resolver = FuzzyMatchResolver(
    driver=driver,
    neo4j_database="neo4j",
    # Optional: filter_query to restrict which nodes to resolve
)
asyncio.run(resolver.run())

Install: pip install neo4j-graphrag[fuzzy] (adds rapidfuzz)

SpacySemanticMatchResolver

Merges entities with same label + semantically similar textual properties via spaCy:

from neo4j_graphrag.experimental.components.resolver import SpacySemanticMatchResolver

resolver = SpacySemanticMatchResolver(driver=driver, neo4j_database="neo4j")
asyncio.run(resolver.run())

Install: pip install neo4j-graphrag[spacy] + python -m spacy download en_core_web_lg

Risk of over-merging ("Apple" company vs "Apple" fruit). Apply domain filter_query to restrict by label.

---

Custom Document Loaders

Extend PdfLoader (pre-process text)

from neo4j_graphrag.experimental.components.pdf_loader import PdfLoader
import re

class CustomPDFLoader(PdfLoader):
    async def run(self, filepath: str):
        doc = await super().run(filepath)
        # Strip AsciiDoc attribute lines
        doc.text = re.sub(r"^:[\w-]+:.*$", "", doc.text, flags=re.MULTILINE)
        return doc

pipeline = SimpleKGPipeline(..., pdf_loader=CustomPDFLoader())

Custom DataLoader (load from any source)

from neo4j_graphrag.experimental.components.pdf_loader import DataLoader, PdfDocument, DocumentInfo

class TextFileLoader(DataLoader):
    async def run(self, filepath: str) -> PdfDocument:
        with open(filepath) as f:
            text = f.read()
        return PdfDocument(
            text=text,
            document_info=DocumentInfo(path=filepath, metadata={"source": "text_file"}),
        )

pipeline = SimpleKGPipeline(..., pdf_loader=TextFileLoader(), from_file=True)
asyncio.run(pipeline.run_async(file_path="data/document.txt"))

---

KG Builder Prompt Customization

Prepend domain instructions

from neo4j_graphrag.experimental.components.entity_relation_extractor import (
    LLMEntityRelationExtractor,
)

# Simple approach: pass domain prefix to pipeline's prompt_template
domain_instructions = (
    "Extract ONLY technology companies, products, and people. "
    "Ignore financial data, dates, and locations."
)
pipeline = SimpleKGPipeline(
    ...,
    prompt_template=domain_instructions,  # prepended to default extraction prompt
)

Full custom extraction prompt

from neo4j_graphrag.experimental.components.entity_relation_extractor import (
    LLMEntityRelationExtractor,
    EntityExtractionPromptTemplate,
)

custom_prompt = EntityExtractionPromptTemplate(
    template="""You are a KG extractor. Extract entities from:
{text}
Schema: {schema}
Output JSON only."""
)
extractor = LLMEntityRelationExtractor(llm=llm, prompt_template=custom_prompt)

LLM adapter for local/custom providers

from neo4j_graphrag.llm import OpenAILLM

# LM Studio local model
llm = OpenAILLM(
    model_name="openai/gpt-oss-20b",
    model_params={"temperature": 0},
    base_url="http://localhost:1234/v1",
)

Custom provider: inherit neo4j_graphrag.llm.base.LLMInterface, implement invoke() and ainvoke().

---

Explore Extracted Graph (Cypher)

// View entities per chunk
MATCH p = (c:Chunk)<-[:FROM_CHUNK]-(e1:__Entity__)-[*1..2]->(e2:__Entity__)
RETURN p

// Count entity types
MATCH (e:__Entity__)
RETURN labels(e) AS types, count(*) AS n ORDER BY n DESC

// Find duplicate entities (pre-resolution check)
MATCH (e:__Entity__)
WITH e.name AS name, labels(e) AS lbl, count(*) AS cnt
WHERE cnt > 1
RETURN name, lbl, cnt ORDER BY cnt DESC

Retriever API Reference

VectorRetriever

VectorRetriever(
    driver,
    index_name: str,
    embedder=None,           # required if passing query_text; optional if passing query_vector
    return_properties: list[str] = None,  # subset of node props to return
    result_formatter=None,   # callable(neo4j.Record) -> RetrieverResultItem
    neo4j_database: str = None,
)

search() params: query_text | query_vector, top_k=5, filters={}, effective_search_ratio=1

VectorCypherRetriever

VectorCypherRetriever(
    driver,
    index_name: str,
    retrieval_query: str,    # Cypher fragment; receives `node` and `score`
    embedder=None,
    result_formatter=None,
    neo4j_database: str = None,
)

search() params: query_text | query_vector, top_k=5, query_params={}, filters={}, effective_search_ratio=1

HybridRetriever

HybridRetriever(
    driver,
    vector_index_name: str,
    fulltext_index_name: str,
    embedder=None,
    return_properties: list[str] = None,
    result_formatter=None,
    neo4j_database: str = None,
)

search() params: query_text (required — used for both vector and fulltext), top_k=5, filters={}, effective_search_ratio=1

HybridCypherRetriever

HybridCypherRetriever(
    driver,
    vector_index_name: str,
    fulltext_index_name: str,
    retrieval_query: str,
    embedder=None,
    result_formatter=None,
    neo4j_database: str = None,
)

search() params: query_text (required), top_k=5, query_params={}, filters={}, effective_search_ratio=1

Text2CypherRetriever

Text2CypherRetriever(
    driver,
    llm,                     # any LLMInterface implementation
    neo4j_schema: str = None,  # None = auto-fetched; pass trimmed string for large schemas
    examples: list[str] = None,  # few-shot examples as "Q: ... A: MATCH ..."
    neo4j_database: str = None,
)

search() params: query_text (natural language question), query_params={}

No embedder needed. LLM generates the Cypher; neo4j_schema is injected into prompt.

ToolsRetriever

ToolsRetriever(
    driver,
    llm,
    tools: list,             # list of retrievers converted via convert_to_tool()
    system_instruction: str = None,
    neo4j_database: str = None,
)

Convert a retriever to a tool:

from neo4j_graphrag.tool import convert_to_tool
tool = convert_to_tool(retriever, name="vector_search", description="Searches by embedding similarity")

External Vector DB Retrievers

from neo4j_graphrag.retrievers import (
    WeaviateNeo4jRetriever,
    PineconeNeo4jRetriever,
    QdrantNeo4jRetriever,
)
# Each maps external vector store IDs to Neo4j node IDs
# Requires: pip install neo4j-graphrag[weaviate|pinecone|qdrant]

result_formatter Pattern

from neo4j_graphrag.types import RetrieverResultItem

def my_formatter(record: neo4j.Record) -> RetrieverResultItem:
    return RetrieverResultItem(
        content=f"[{record['article_title']}] {record['chunk_text']}",
        metadata={"orgs": record["mentioned_organizations"], "score": record["score"]},
    )

retriever = VectorCypherRetriever(..., result_formatter=my_formatter)

effective_search_ratio

Controls candidate pool: candidates = top_k * effective_search_ratio. Increase (e.g. 2–5) when retrieval_query filters reduce results significantly.

Pre-filter Operators

filters = {
    "property_name": {"$eq": value},
    "property_name": {"$in": [v1, v2]},
    "property_name": {"$between": {"min": 0, "max": 1}},
    "property_name": {"$like": "prefix%"},   # case-sensitive
    "property_name": {"$ilike": "prefix%"},  # case-insensitive
}
# Combine: {"$and": [{...}, {...}]} or {"$or": [{...}, {...}]}