Neo4j Python Tutorial: Driver, Transactions, GraphRAG, and LLM Integration for 2026

TL;DR — Neo4j Python tutorial: driver, transactions, GraphRAG. Neo4j GraphRAG Python: "Official Neo4j GraphRAG features for Python — create_vector_index, retrievers, LLM integration." Getting Started: "neo4j-graphrag: OpenAI LLM, LangChain compatible, custom retrievers, configurable components." RAG User Guide: "HybridCypherRetriever: vector + fulltext, graph traversal for context. pip install neo4j_graphrag[openai]." Driver Best Practices: "Driver lifecycle, connection pooling, session management, managed transactions." Learn more with Cypher tutorial, build from documents, GraphRAG tutorial, and FalkorDB vs Neo4j.

Neo4j GraphRAG Python introduces the package: "This package contains the official Neo4j GraphRAG features for Python — from neo4j import GraphDatabase, from neo4j_graphrag.indexes import create_vector_index."

Getting Started adds: "The neo4j-graphrag package provides an implementation for OpenAI LLMs, but its interface is compatible with LangChain chat models and lets you write your own interface if needed. A retriever — the package provides implementations and lets you write your own."

Neo4j Python Architecture

flowchart TD subgraph Connection["Driver Connection"] Driver["GraphDatabase.driver
bolt://localhost:7687
auth=('neo4j', 'password')
connection pool (automatic)
singleton per application"] Session["Session Management
with driver.session() as sess
context manager
auto-close
one per transaction group"] Async["Async Driver
AsyncGraphDatabase.driver
for async frameworks
FastAPI, asyncio"] end subgraph Transactions["Transaction Types"] Managed["Managed Transactions
session.execute_write(tx_fn)
session.execute_read(tx_fn)
auto-retry on transient errors
atomic — all or nothing"] AutoCommit["Auto-Commit
session.run(query, params)
single query, no retry
simple operations"] Batch["Batch Operations
UNWIND $list AS item
MERGE in bulk
efficient for ingestion"] end subgraph GraphRAG["neo4j-graphrag Package"] Index["Vector Index
create_vector_index()
dimensions, similarity
cosine, Euclidean"] Retrievers["Retrievers
VectorRetriever
HybridCypherRetriever
vector + fulltext + graph
custom retriever support"] LLM["LLM Integration
OpenAI (built-in)
LangChain compatible
custom interface
configurable prompt"] end subgraph Frameworks["Framework Integration"] LangChain["LangChain
Neo4jGraph
LLMGraphTransformer
GraphCypherQAChain
Neo4jVector"] LlamaIndex["LlamaIndex
Neo4jPropertyGraphStore
SchemaLLMPathExtractor
LLMSynonymRetriever
CypherTemplateRetriever"] end subgraph Production["Production Best Practices"] Singleton["Singleton Driver
one per application
connection pool managed
driver.close() on shutdown"] Params["Parameterized Queries
$param syntax
prevents injection
type-safe"] Pool["Connection Pool
max_connection_pool_size
connection_timeout
max_connection_lifetime
neo4j:// for cluster routing"] Monitor["Monitoring
driver metrics
query logging
performance profiling
index frequently queried props"] end Driver --> Session Driver --> Async Session --> Managed Session --> AutoCommit Session --> Batch Index --> Retrievers Retrievers --> LLM Driver --> LangChain Driver --> LlamaIndex Driver --> Singleton Session --> Params Driver --> Pool Driver --> Monitor style Connection fill:#4169E1,color:#fff style Transactions fill:#39FF14,color:#000 style GraphRAG fill:#2D1B69,color:#fff style Production fill:#FF6B6B,color:#fff

Integration Comparison

Integration Package Key Classes Best For
neo4j driver neo4j GraphDatabase, Session Direct Cypher queries
neo4j-graphrag neo4j-graphrag[openai] HybridCypherRetriever, create_vector_index GraphRAG pipeline
LangChain langchain-community Neo4jGraph, LLMGraphTransformer, GraphCypherQAChain LangChain ecosystem
LlamaIndex llama-index-graph-stores-neo4j Neo4jPropertyGraphStore, SchemaLLMPathExtractor LlamaIndex ecosystem

Implementation

from dataclasses import dataclass
from typing import Optional
from enum import Enum

class IntegrationType(Enum):
    DRIVER = "driver"
    GRAPHRAG = "graphrag"
    LANGCHAIN = "langchain"
    LLAMAINDEX = "llamaindex"

@dataclass
class Neo4jPythonGuide:
    """Neo4j Python tutorial implementation guide."""

    def get_driver_basics(self) -> str:
        """Neo4j Python driver basics."""
        return (
            "# === NEO4J DRIVER BASICS ===\n"
            "# pip install neo4j\n"
            "from neo4j import GraphDatabase\n"
            "\n"
            "# 1. CREATE DRIVER (singleton)\n"
            "driver = GraphDatabase.driver(\n"
            "    'bolt://localhost:7687',\n"
            "    auth=('neo4j', 'password'),\n"
            "    # Connection pool settings\n"
            "    max_connection_pool_size=50,\n"
            "    connection_timeout=30,\n"
            "    max_connection_lifetime=3600,\n"
            ")\n"
            "\n"
            "# 2. SIMPLE QUERY\n"
            "with driver.session() as sess:\n"
            "    result = sess.run(\n"
            "        '''MATCH (p:Person)\n"
            "          -[:WORKS_AT]->(o)\n"
            "          RETURN p.name, o.name''')\n"
            "    for record in result:\n"
            "        print(\n"
            "            f'{record[\"p.name\"]} '\n"
            "            f'works at '\n"
            "            f'{record[\"o.name\"]}')\n"
            "\n"
            "# 3. PARAMETERIZED QUERY\n"
            "with driver.session() as sess:\n"
            "    result = sess.run(\n"
            "        '''MATCH (p:Person)\n"
            "          WHERE p.name = $name\n"
            "          RETURN p''',\n"
            "        name='Dr. Smith')\n"
            "    for record in result:\n"
            "        print(record['p'])\n"
            "\n"
            "# 4. VERIFY CONNECTION\n"
            "driver.verify_connectivity()\n"
            "print('Connected!')\n"
            "\n"
            "# 5. CLOSE DRIVER\n"
            "driver.close()"
        )

    def get_transactions(self) -> str:
        """Transaction management."""
        return (
            "# === TRANSACTIONS ===\n"
            "\n"
            "# 1. MANAGED WRITE TRANSACTION\n"
            "def add_person(tx, name, role):\n"
            "    result = tx.run(\n"
            "        '''MERGE (p:Person\n"
            "          {name: $name})\n"
            "          SET p.role = $role\n"
            "          RETURN p''',\n"
            "        name=name, role=role)\n"
            "    return result.single()\n"
            "\n"
            "with driver.session() as sess:\n"
            "    record = sess.execute_write(\n"
            "        add_person,\n"
            "        'Alice', 'Engineer')\n"
            "    print(record['p'])\n"
            "\n"
            "# 2. MANAGED READ TRANSACTION\n"
            "def get_colleagues(tx, org_name):\n"
            "    result = tx.run(\n"
            "        '''MATCH (p:Person)\n"
            "          -[:WORKS_AT]->\n"
            "          (o:Organization\n"
            "           {name: $org})\n"
            "          RETURN p.name''',\n"
            "        org=org_name)\n"
            "    return [r['p.name']\n"
            "            for r in result]\n"
            "\n"
            "with driver.session() as sess:\n"
            "    names = sess.execute_read(\n"
            "        get_colleagues, 'AI Lab')\n"
            "    print(names)\n"
            "\n"
            "# 3. BATCH WITH UNWIND\n"
            "def batch_add(tx, people):\n"
            "    tx.run(\n"
            "        '''UNWIND $people AS person\n"
            "          MERGE (p:Person\n"
            "            {name: person.name})\n"
            "          SET p.role = person.role\n"
            "          MERGE (o:Organization\n"
            "            {name: person.org})\n"
            "          MERGE (p)-[:WORKS_AT]->(o)''',\n"
            "        people=people)\n"
            "\n"
            "with driver.session() as sess:\n"
            "    sess.execute_write(\n"
            "        batch_add,\n"
            "        people=[\n"
            "            {'name':'Alice',\n"
            "             'role':'Engineer',\n"
            "             'org':'AI Lab'},\n"
            "            {'name':'Bob',\n"
            "             'role':'Designer',\n"
            "             'org':'AI Lab'},\n"
            "        ])\n"
            "\n"
            "# 4. MULTI-QUERY TRANSACTION\n"
            "def create_with_relations(tx):\n"
            "    tx.run(\n"
            "        'MERGE (p:Person'\n"
            "        ' {name:$name})',\n"
            "        name='Carol')\n"
            "    tx.run(\n"
            "        'MERGE (o:Organization'\n"
            "        ' {name:$org})',\n"
            "        org='TechCorp')\n"
            "    tx.run(\n"
            "        '''MATCH (p:Person\n"
            "          {name:$name}),\n"
            "          (o:Organization\n"
            "           {name:$org})\n"
            "          MERGE (p)-[:WORKS_AT]->(o)''',\n"
            "        name='Carol',\n"
            "        org='TechCorp')\n"
            "\n"
            "with driver.session() as sess:\n"
            "    sess.execute_write(\n"
            "        create_with_relations)"
        )

    def get_graphrag_package(self) -> str:
        """neo4j-graphrag package usage."""
        return (
            "# === NEO4J-GRAPHRAG PACKAGE ===\n"
            "# pip install neo4j-graphrag[openai]\n"
            "\n"
            "from neo4j import GraphDatabase\n"
            "from neo4j_graphrag.indexes\n"
            "    import create_vector_index\n"
            "from neo4j_graphrag.retrievers\n"
            "    import (\n"
            "    VectorRetriever,\n"
            "    HybridCypherRetriever,\n"
            ")\n"
            "from neo4j_graphrag.llm\n"
            "    import OpenAILLM\n"
            "from neo4j_graphrag.generation\n"
            "    import GraphRAG\n"
            "\n"
            "# 1. CONNECT\n"
            "URI = 'neo4j://localhost:7687'\n"
            "AUTH = ('neo4j', 'password')\n"
            "driver = GraphDatabase.driver(\n"
            "    URI, auth=AUTH)\n"
            "\n"
            "# 2. CREATE VECTOR INDEX\n"
            "create_vector_index(\n"
            "    driver,\n"
            "    'chunk_embeddings',\n"
            "    label='Chunk',\n"
            "    embedding_property=\n"
            "        'embedding',\n"
            "    dimensions=384,\n"
            "    similarity_fn='cosine',\n"
            ")\n"
            "\n"
            "# 3. VECTOR RETRIEVER\n"
            "retriever = VectorRetriever(\n"
            "    driver=driver,\n"
            "    index_name=\n"
            "        'chunk_embeddings',\n"
            "    embedder=embedder,\n"
            ")\n"
            "results = retriever.search(\n"
            "    query_text='Who works at AI Lab?',\n"
            "    top_k=5,\n"
            ")\n"
            "\n"
            "# 4. HYBRID CYPHER RETRIEVER\n"
            "# vector + fulltext + graph\n"
            "retrieval_query = '''\n"
            "    MATCH (node)-[:MENTIONS]\n"
            "      ->(e)\n"
            "    MATCH (e)<-[:MENTIONS]-\n"
            "      (c2:Chunk)\n"
            "    RETURN c2.text AS text,\n"
            "      collect(e.name) AS entities\n"
            "'''\n"
            "hybrid = HybridCypherRetriever(\n"
            "    driver=driver,\n"
            "    index_name=\n"
            "        'chunk_embeddings',\n"
            "    fulltext_index_name=\n"
            "        'chunk_text',\n"
            "    retrieval_query=retrieval_query,\n"
            ")\n"
            "results = hybrid.search(\n"
            "    query_text='AI Lab employees',\n"
            "    query_vector=embedding,\n"
            "    top_k=5,\n"
            ")\n"
            "\n"
            "# 5. LLM + GraphRAG\n"
            "llm = OpenAILLM(\n"
            "    model_name='gpt-4o',\n"
            "    api_key='...')\n"
            "rag = GraphRAG(\n"
            "    retriever=hybrid,\n"
            "    llm=llm,\n"
            ")\n"
            "result = rag.search(\n"
            "    'Who works at AI Lab?')\n"
            "print(result.answer)"
        )

    def get_langchain_integration(self) -> str:
        """LangChain Neo4j integration."""
        return (
            "# === LANGCHAIN INTEGRATION ===\n"
            "# pip install langchain-community\n"
            "#   langchain-experimental\n"
            "\n"
            "from langchain_community.graphs\n"
            "    import Neo4jGraph\n"
            "from langchain_experimental.\n"
            "    graph_transformers import\n"
            "    LLMGraphTransformer\n"
            "from langchain.chains\n"
            "    import GraphCypherQAChain\n"
            "from langchain_openai\n"
            "    import ChatOpenAI\n"
            "\n"
            "# 1. CONNECT\n"
            "graph = Neo4jGraph(\n"
            "    url='bolt://localhost:7687',\n"
            "    username='neo4j',\n"
            "    password='password',\n"
            ")\n"
            "graph.refresh_schema()\n"
            "print(graph.schema)\n"
            "\n"
            "# 2. QUERY\n"
            "result = graph.query(\n"
            "    '''MATCH (p:Person)\n"
            "      -[:WORKS_AT]->(o)\n"
            "      RETURN p.name, o.name''')\n"
            "\n"
            "# 3. LLM GRAPH TRANSFORMER\n"
            "llm = ChatOpenAI(\n"
            "    base_url=\n"
            "      'http://localhost:11434/v1',\n"
            "    api_key='ollama',\n"
            "    model='llama3.2')\n"
            "transformer = (\n"
            "    LLMGraphTransformer(\n"
            "    llm=llm,\n"
            "    allowed_nodes=['Person',\n"
            "      'Organization'],\n"
            "    allowed_relationships=[\n"
            "      'WORKS_AT']))\n"
            "graph_docs = (\n"
            "    transformer\n"
            "    .convert_to_graph_documents(\n"
            "    documents))\n"
            "graph.add_graph_documents(\n"
            "    graph_docs)\n"
            "\n"
            "# 4. NL TO CYPHER\n"
            "chain = (\n"
            "    GraphCypherQAChain\n"
            "    .from_llm(\n"
            "    llm=llm, graph=graph,\n"
            "    verbose=True))\n"
            "result = chain.invoke({\n"
            "    'query': 'Who works at'\n"
            "      ' AI Lab?'})\n"
            "print(result['result'])"
        )

    def get_llamaindex_integration(self) -> str:
        """LlamaIndex Neo4j integration."""
        return (
            "# === LLAMAINDEX INTEGRATION ===\n"
            "# pip install llama-index-\n"
            "#   graph-stores-neo4j\n"
            "\n"
            "from llama_index.core import\n"
            "    PropertyGraphIndex,\n"
            "    SimpleDirectoryReader\n"
            "from llama_index.core.indices.\n"
            "    property_graph import (\n"
            "    SchemaLLMPathExtractor,\n"
            "    LLMSynonymRetriever,\n"
            "    VectorContextRetriever,\n"
            "    CypherTemplateRetriever,\n"
            ")\n"
            "from llama_index.graph_stores.\n"
            "    neo4j import\n"
            "    Neo4jPropertyGraphStore\n"
            "from llama_index.llms.ollama\n"
            "    import Ollama\n"
            "from typing import Literal\n"
            "\n"
            "# 1. GRAPH STORE\n"
            "graph_store = (\n"
            "    Neo4jPropertyGraphStore(\n"
            "    username='neo4j',\n"
            "    password='password',\n"
            "    url='bolt://localhost:7687'))\n"
            "\n"
            "# 2. SCHEMA\n"
            "entities = Literal[\n"
            "    'PERSON', 'ORGANIZATION',\n"
            "    'TOPIC']\n"
            "relations = Literal[\n"
            "    'WORKS_AT', 'MENTIONS',\n"
            "    'CITES']\n"
            "validation = {\n"
            "    'PERSON': ['WORKS_AT'],\n"
            "    'ORGANIZATION': [],\n"
            "    'TOPIC': ['MENTIONS'],\n"
            "}\n"
            "extractor = (\n"
            "    SchemaLLMPathExtractor(\n"
            "    llm=Ollama(model='llama3.2',\n"
            "        json_mode=True),\n"
            "    possible_entities=entities,\n"
            "    possible_relations=relations,\n"
            "    kg_validation_schema=\n"
            "        validation,\n"
            "    strict=True,\n"
            "    num_workers=4))\n"
            "\n"
            "# 3. BUILD INDEX\n"
            "docs = SimpleDirectoryReader(\n"
            "    './docs').load_data()\n"
            "index = (\n"
            "    PropertyGraphIndex\n"
            "    .from_documents(\n"
            "    docs,\n"
            "    kg_extractors=[extractor],\n"
            "    property_graph_store=\n"
            "        graph_store,\n"
            "    show_progress=True))\n"
            "\n"
            "# 4. HYBRID RETRIEVAL\n"
            "retriever = index.as_retriever(\n"
            "    sub_retrievers=[\n"
            "        LLMSynonymRetriever(\n"
            "            index.property_graph_store,\n"
            "            llm=Ollama(\n"
            "              model='llama3.2'),\n"
            "            include_text=True),\n"
            "        VectorContextRetriever(\n"
            "            index.property_graph_store,\n"
            "            include_text=True),\n"
            "        CypherTemplateRetriever(\n"
            "            index.property_graph_store),\n"
            "    ])\n"
            "nodes = retriever.retrieve(\n"
            "    'Who works at AI Lab?')"
        )

    def get_async_driver(self) -> str:
        """Async Neo4j driver."""
        return (
            "# === ASYNC DRIVER ===\n"
            "from neo4j import (\n"
            "    AsyncGraphDatabase)\n"
            "import asyncio\n"
            "\n"
            "async def main():\n"
            "    driver = (\n"
            "        AsyncGraphDatabase.driver(\n"
            "        'bolt://localhost:7687',\n"
            "        auth=('neo4j',\n"
            "          'password')))\n"
            "    \n"
            "    async with driver.session() as sess:\n"
            "        result = await sess.run(\n"
            "            '''MATCH (p:Person)\n"
            "              RETURN p.name''')\n"
            "        records = [\n"
            "            r async for r in result]\n"
            "        for r in records:\n"
            "            print(r['p.name'])\n"
            "    \n"
            "    await driver.close()\n"
            "\n"
            "asyncio.run(main())"
        )

    def get_best_practices(self) -> dict:
        """Production best practices."""
        return {
            "driver_lifecycle": "One driver per application (singleton). Driver manages connection pool. driver.close() on shutdown.",
            "session_management": "Use context managers (with) for sessions. One session per transaction group. Auto-close.",
            "parameterized_queries": "Always use $param syntax. Prevents Cypher injection. Type-safe.",
            "managed_transactions": "Use execute_write/execute_read for auto-retry on transient errors. Atomic — all or nothing.",
            "batch_operations": "Use UNWIND for bulk operations. More efficient than individual queries.",
            "connection_pool": "Configure max_connection_pool_size for concurrent workloads. Set connection_timeout and max_connection_lifetime.",
            "cluster_routing": "Use neo4j:// scheme for routing in cluster deployments. bolt:// for single instance.",
            "async": "Use AsyncGraphDatabase.driver for async frameworks (FastAPI, asyncio).",
            "indexing": "Create indexes on frequently queried properties. Use constraints for unique properties.",
            "monitoring": "Monitor driver metrics. Log queries for debugging. Profile slow queries.",
        }

Neo4j Python Checklist

  • [ ] Neo4j Python driver: pip install neo4j, from neo4j import GraphDatabase
  • [ ] Create driver: GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password'))
  • [ ] Driver is singleton — one per application, manages connection pool automatically
  • [ ] Use context managers: with driver.session() as sess: for auto-close
  • [ ] Parameterized queries: $param syntax prevents Cypher injection
  • [ ] driver.verify_connectivity() to test connection
  • [ ] driver.close() on application shutdown
  • [ ] Connection pool settings: max_connection_pool_size, connection_timeout, max_connection_lifetime
  • [ ] Use neo4j:// scheme for cluster routing, bolt:// for single instance
  • [ ] Managed transactions: session.execute_write(tx_fn) for writes, session.execute_read(tx_fn) for reads
  • [ ] Managed transactions are auto-retried on transient errors — preferred for production
  • [ ] Transaction functions: def tx_fn(tx, params): tx.run(query, params)
  • [ ] Auto-commit: session.run(query) for single queries (no retry, no atomicity guarantee)
  • [ ] Batch operations: UNWIND $list AS item for bulk MERGE/CREATE
  • [ ] Multi-query transactions: chain tx.run() calls in same transaction function — atomic
  • [ ] Async driver: AsyncGraphDatabase.driver() for FastAPI/asyncio — async with driver.session()
  • [ ] neo4j-graphrag package: pip install neo4j-graphrag[openai]
  • [ ] neo4j-graphrag: from neo4j_graphrag.indexes import create_vector_index
  • [ ] neo4j-graphrag: from neo4j_graphrag.retrievers import VectorRetriever, HybridCypherRetriever
  • [ ] neo4j-graphrag: from neo4j_graphrag.llm import OpenAILLM, from neo4j_graphrag.generation import GraphRAG
  • [ ] neo4j-graphrag: OpenAI LLM built-in, LangChain chat models compatible, custom interface supported
  • [ ] neo4j-graphrag: HybridCypherRetriever — vector + fulltext index + graph traversal retrieval query
  • [ ] neo4j-graphrag: each component configurable — LLM, prompt, retriever
  • [ ] neo4j-graphrag: GraphRAG(retriever, llm).search(query) for end-to-end RAG
  • [ ] LangChain: from langchain_community.graphs import Neo4jGraph
  • [ ] LangChain: Neo4jGraph(url, username, password), graph.query(), graph.refresh_schema()
  • [ ] LangChain: LLMGraphTransformer for entity extraction from documents
  • [ ] LangChain: GraphCypherQAChain.from_llm() for natural language to Cypher
  • [ ] LangChain: graph.add_graph_documents() for ingestion
  • [ ] LlamaIndex: from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore
  • [ ] LlamaIndex: Neo4jPropertyGraphStore(username, password, url)
  • [ ] LlamaIndex: PropertyGraphIndex.from_documents(docs, kg_extractors, property_graph_store)
  • [ ] LlamaIndex: SchemaLLMPathExtractor with Literal entities/relations, validation_schema, strict
  • [ ] LlamaIndex: LLMSynonymRetriever, VectorContextRetriever, CypherTemplateRetriever
  • [ ] LlamaIndex: combine all three retrievers for hybrid search
  • [ ] Index frequently queried properties for performance
  • [ ] Create constraints for unique properties (MERGE performance)
  • [ ] Create vector indexes for embedding similarity search
  • [ ] Monitor driver metrics and log queries for debugging
  • [ ] Profile slow queries with EXPLAIN and PROFILE
  • [ ] Read Cypher tutorial for query language
  • [ ] Read build from documents for ingestion
  • [ ] Read GraphRAG tutorial for GraphRAG frameworks
  • [ ] Read FalkorDB vs Neo4j for database comparison
  • [ ] Test: driver connects and queries return expected results
  • [ ] Test: managed transactions are atomic and auto-retried
  • [ ] Test: parameterized queries prevent injection
  • [ ] Test: batch UNWIND performance for bulk operations
  • [ ] Test: async driver works with your async framework
  • [ ] Test: neo4j-graphrag retriever returns relevant results
  • [ ] Test: LangChain/LlamaIndex integration works end-to-end
  • [ ] Test: connection pool handles concurrent workloads
  • [ ] Document driver config, transaction patterns, GraphRAG setup, framework integration

FAQ

How do you connect to Neo4j from Python?

Use the neo4j Python driver with GraphDatabase.driver and session-based queries. Neo4j Driver Best Practices: "GraphDatabase.driver for connection, session for queries, driver lifecycle management." Connection: (1) from neo4j import GraphDatabase. (2) driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password')). (3) with driver.session() as session: session.run('Cypher query', params). (4) Parameterized queries with $param for safety. (5) driver.close() when done. (6) Connection pooling is automatic. (7) Neo4j 5.26+ supports bolt:// and neo4j:// schemes. (8) Auth can be basic, Kerberos, or custom. (9) For async: AsyncGraphDatabase.driver. (10) Always use context manager (with) for sessions.

How do you use transactions in the Neo4j Python driver?

Use session.execute_write and session.execute_read for managed transactions. Transactions: (1) session.execute_write(tx_function) for write transactions. (2) session.execute_read(tx_function) for read transactions. (3) tx.run(query, params) inside transaction function. (4) Transaction functions are auto-retried on transient errors. (5) def add_person(tx, name): tx.run('MERGE (p:Person {name:$name})', name=name). (6) session.execute_write(add_person, 'Alice'). (7) Return values from tx.run can be consumed in the function. (8) For multiple queries in one transaction, chain tx.run calls. (9) Transactions are atomic — all succeed or all fail. (10) Use for batch operations with UNWIND for efficiency. (11) Auto-commit: session.run() for single queries (no retry). (12) Managed transactions preferred for production.

How do you use the neo4j-graphrag Python package?

neo4j-graphrag provides retrievers, vector indexes, and LLM integration for GraphRAG. Neo4j GraphRAG Python: "Official Neo4j GraphRAG features for Python. create_vector_index, retrievers, LLM integration." Getting Started: "neo4j-graphrag package provides OpenAI LLM implementation, compatible with LangChain chat models, write your own interface. Retriever implementations and custom retriever support." RAG User Guide: "pip install neo4j_graphrag[openai]. HybridCypherRetriever: vector + fulltext index, retrieval query traverses graph. Each component configurable: LLM, prompt, retriever." Usage: (1) pip install neo4j-graphrag[openai]. (2) from neo4j_graphrag.indexes import create_vector_index. (3) from neo4j_graphrag.retrievers import HybridCypherRetriever. (4) Configure retriever with index names and retrieval query. (5) retriever.search(query_vector=embedding, top_k=5). (6) Custom LLM interface or LangChain chat models.

How do you integrate Neo4j with LangChain and LlamaIndex?

Both frameworks have first-class Neo4j integrations for GraphRAG. LangChain: (1) from langchain_community.graphs import Neo4jGraph. (2) graph = Neo4jGraph(url, username, password). (3) graph.query('Cypher query'). (4) LLMGraphTransformer for entity extraction. (5) GraphCypherQAChain for NL to Cypher. (6) Neo4jVector for vector store. LlamaIndex: (1) from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore. (2) graph_store = Neo4jPropertyGraphStore(username, password, url). (3) PropertyGraphIndex.from_documents with kg_extractors. (4) SchemaLLMPathExtractor for schema-constrained extraction. (5) LLMSynonymRetriever, VectorContextRetriever, CypherTemplateRetriever. (6) Combine retrievers for hybrid search. Both: Cypher query language, vector index support, Python drivers, GraphRAG pipeline.

What are Neo4j Python driver best practices for production?

Manage driver lifecycle, use connection pooling, parameterized queries, and managed transactions. Neo4j Driver Best Practices: "Driver lifecycle management, connection pooling, session management." Best practices: (1) One driver per application (singleton) — driver manages connection pool. (2) Use context managers (with) for sessions. (3) Parameterized queries ($param) to prevent injection. (4) Managed transactions (execute_write/execute_read) for auto-retry. (5) Batch with UNWIND for bulk operations. (6) driver.close() on application shutdown. (7) Configure max_connection_pool_size for concurrent workloads. (8) Use neo4j:// scheme for routing in cluster. (9) Set connection_timeout and max_connection_lifetime. (10) Monitor driver metrics. (11) Use async driver (AsyncGraphDatabase) for async frameworks. (12) Index frequently queried properties for performance.


Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →