neo4j-graphrag-skill

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

// 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'

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()

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},
)

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?")

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()

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

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"}},
)

# 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,
)

# v1.19+ — preferred
from neo4j_graphrag.components.text_splitters.fixed_size_splitter import FixedSizeSplitter
 
# deprecated (removed in 2.0)
from neo4j_graphrag.experimental.components.text_splitters.fixed_size_splitter import FixedSizeSplitter
 
# SimpleKGPipeline did NOT move — only valid path:
from neo4j_graphrag.experimental.pipeline.kg_builder import SimpleKGPipeline

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

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)

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

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)

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]

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