Query a knowledge graph in LlamaIndex with n4n.ai

This llamaindex knowledge graph query n4n.ai tutorial walks you through building a production-ready knowledge graph query pipeline. You’ll extract entities and relationships from documents, store them in a graph database, and query the graph using natural language — all routed through n4n.ai’s unified endpoint for model access and automatic fallback.

Step 1: Set up the environment and dependencies

Start with a clean virtual environment. You need LlamaIndex core, the knowledge graph modules, a graph store (we’ll use Neo4j for this tutorial), and the OpenAI-compatible client for n4n.ai.

python -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install \
  llama-index \
  llama-index-graph-stores-neo4j \
  llama-index-llms-openai \
  llama-index-embeddings-openai \
  neo4j \
  python-dotenv

Create a .env file with your credentials. The n4n.ai endpoint uses the OpenAI API format, so you only need to point the base URL and provide your n4n.ai key.

# .env
N4N_API_KEY=your_n4n_key_here
N4N_BASE_URL=https://api.n4n.ai/v1
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=your_neo4j_password

Verify the environment loads correctly:

# verify_env.py
import os
from dotenv import load_dotenv

load_dotenv()

required = ["N4N_API_KEY", "N4N_BASE_URL", "NEO4J_URI", "NEO4J_USER", "NEO4J_PASSWORD"]
for var in required:
    val = os.getenv(var)
    if not val:
        raise RuntimeError(f"Missing {var}")
    print(f"{var}: {'*' * 8}{val[-4:] if len(val) > 4 else '***'}")

print("Environment OK")

Run it:

python verify_env.py

You should see all four variables masked with their last four characters.

Step 2: Configure the n4n.ai LLM and embedding models

LlamaIndex’s OpenAI and OpenAIEmbedding classes accept api_base and api_key parameters, making n4n.ai a drop-in replacement. Configure both the LLM for extraction and querying, and the embedding model for any vector fallback.

# config.py
import os
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import Settings

load_dotenv()

llm = OpenAI(
    model="meta-llama/llama-3.1-70b-instruct",
    api_base=os.getenv("N4N_BASE_URL"),
    api_key=os.getenv("N4N_API_KEY"),
    temperature=0.0,
    max_tokens=2048,
)

embed_model = OpenAIEmbedding(
    model="text-embedding-3-small",
    api_base=os.getenv("N4N_BASE_URL"),
    api_key=os.getenv("N4N_API_KEY"),
)

Settings.llm = llm
Settings.embed_model = embed_model
Settings.chunk_size = 512
Settings.chunk_overlap = 50

The model string "meta-llama/llama-3.1-70b-instruct" is an n4n.ai routing identifier. n4n.ai resolves this to an available provider (Together, Fireworks, etc.) and handles fallback automatically if the primary provider is rate-limited or degraded.

Test the configuration:

# test_config.py
from config import llm, embed_model

resp = llm.complete("Return only the word 'ok'")
print(f"LLM test: {resp.text.strip()}")

emb = embed_model.get_text_embedding("test")
print(f"Embedding dim: {len(emb)}")

Run it. You should see LLM test: ok and Embedding dim: 1536.

Step 3: Start Neo4j and create the graph store

If you don’t have Neo4j running, the fastest path is Docker:

docker run -d \
  --name neo4j \
  -p 7474:7474 -p 7687:7687 \
  -e NEO4J_AUTH=neo4j/your_neo4j_password \
  neo4j:5.23

Wait 15-20 seconds for startup, then verify:

curl -u neo4j:your_neo4j_password http://localhost:7474/db/data/

Now create the LlamaIndex graph store wrapper:

# graph_store.py
import os
from llama_index.graph_stores.neo4j import Neo4jGraphStore
from llama_index.core import StorageContext

graph_store = Neo4jGraphStore(
    username=os.getenv("NEO4J_USER"),
    password=os.getenv("NEO4J_PASSWORD"),
    url=os.getenv("NEO4J_URI"),
    database="neo4j",
)

storage_context = StorageContext.from_defaults(graph_store=graph_store)

Verify connectivity:

# test_graph_store.py
from graph_store import graph_store

# Simple write/read cycle
graph_store.query("CREATE (n:Test {value: 'hello'}) RETURN n")
result = graph_store.query("MATCH (n:Test) RETURN n.value AS val")
print(f"Graph store test: {result}")

# Clean up
graph_store.query("MATCH (n:Test) DETACH DELETE n")

Run it. You should see the test node returned and then cleaned up.

Step 4: Prepare source documents

For this tutorial, use a small set of domain-specific documents. In practice, you’d load from PDFs, Notion, Confluence, or your internal wiki. Here we’ll create a few in-memory Document objects representing technical specifications.

# documents.py
from llama_index.core import Document

docs = [
    Document(
        text=(
            "The n4n.ai gateway routes requests to 240+ models through a single "
            "OpenAI-compatible endpoint. It implements automatic fallback when a "
            "provider is rate-limited or degraded, and forwards provider cache-control "
            "hints to clients. Usage is metered per token."
        ),
        metadata={"source": "architecture-overview.md", "domain": "inference-gateway"},
    ),
    Document(
        text=(
            "LlamaIndex knowledge graph construction uses an LLM to extract entities "
            "and relationships from text. The default schema extracts (subject, "
            "predicate, object) triples. Custom schemas can constrain entity types "
            "and relation types for domain-specific graphs."
        ),
        metadata={"source": "kg-construction.md", "domain": "llamaindex"},
    ),
    Document(
        text=(
            "The KnowledgeGraphIndex in LlamaIndex stores triples in a graph store "
            "and optionally maintains a vector index for hybrid retrieval. Query "
            "engines include KGTableRetriever for structured queries and "
            "KnowledgeGraphQueryEngine for natural language to Cypher translation."
        ),
        metadata={"source": "kg-index.md", "domain": "llamaindex"},
    ),
    Document(
        text=(
            "n4n.ai honors client routing directives such as 'provider: together' "
            "or 'max_latency_ms: 500'. These are passed as extra headers and "
            "respected by the gateway's router. This lets applications steer "
            "traffic without changing model identifiers."
        ),
        metadata={"source": "routing-directives.md", "domain": "inference-gateway"},
    ),
]

Step 5: Build the knowledge graph index

The KnowledgeGraphIndex extracts triples using the configured LLM and writes them to the graph store. We’ll use the default simple schema (no type constraints) for breadth, but you can pass kg_triple_extract_template for custom extraction.

# build_index.py
from llama_index.core import KnowledgeGraphIndex
from config import storage_context
from documents import docs

index = KnowledgeGraphIndex.from_documents(
    docs,
    storage_context=storage_context,
    max_triplets_per_chunk=10,
    include_embeddings=True,  # enables hybrid vector + graph retrieval
    show_progress=True,
)

print(f"Index built with {len(index.index_struct.triplets)} triplets")

Run it. You should see something like Index built with 42 triplets (exact count varies by LLM extraction).

Verify the graph in Neo4j Browser at http://localhost:7474. Run:

MATCH (n)-[r]->(m) RETURN n, r, m LIMIT 50

You should see a network of entities and relationships extracted from your documents.

Step 6: Create the query engine

LlamaIndex provides two main query engines for knowledge graphs. KnowledgeGraphQueryEngine translates natural language to Cypher using the LLM, then executes it. KGTableRetriever retrieves relevant triplets and synthesizes an answer. We’ll use the natural language engine for this tutorial.

# query_engine.py
from llama_index.core.query_engine import KnowledgeGraphQueryEngine
from build_index import index

query_engine = KnowledgeGraphQueryEngine(
    storage_context=index.storage_context,
    llm=index.llm,
    verbose=True,
    # Optional: limit Cypher complexity
    # cypher_query_limit=10,
)

Test with a straightforward question:

# test_query.py
from query_engine import query_engine

questions = [
    "What models does n4n.ai route to?",
    "How does LlamaIndex extract knowledge graph triplets?",
    "What routing directives does n4n.ai support?",
]

for q in questions:
    print(f"\nQ: {q}")
    response = query_engine.query(q)
    print(f"A: {response.response}")
    if hasattr(response, "metadata") and "cypher_query" in response.metadata:
        print(f"Cypher: {response.metadata['cypher_query']}")

Run it. You should see answers grounded in your documents, plus the generated Cypher queries in the metadata.

Example output:

Q: What models does n4n.ai route to?
A: n4n.ai routes requests to 240+ models through a single OpenAI-compatible endpoint.
Cypher: MATCH (n:Entity)-[r:RELATES_TO]->(m:Entity) WHERE n.name CONTAINS 'n4n.ai' OR m.name CONTAINS 'n4n.ai' RETURN n, r, m LIMIT 10

Step 7: Add hybrid retrieval for better coverage

Pure graph queries miss information that wasn’t extracted as clean triples. Enable hybrid mode by combining the graph retriever with a vector index over the same documents.

# hybrid_engine.py
from llama_index.core import VectorStoreIndex
from llama_index.core.retrievers import KnowledgeGraphRAGRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import SimilarityPostprocessor
from build_index import index
from documents import docs

# Vector index over the same docs
vector_index = VectorStoreIndex.from_documents(docs, storage_context=index.storage_context)

# Graph retriever (includes embeddings for hybrid)
graph_retriever = KnowledgeGraphRAGRetriever(
    storage_context=index.storage_context,
    llm=index.llm,
    verbose=True,
    include_text=True,  # returns source text chunks alongside triplets
)

# Vector retriever
vector_retriever = vector_index.as_retriever(similarity_top_k=4)

# Fuse results: simple concatenation + dedup by node_id
class HybridRetriever:
    def __init__(self, graph_retriever, vector_retriever):
        self.graph_retriever = graph_retriever
        self.vector_retriever = vector_retriever

    def retrieve(self, query):
        graph_nodes = self.graph_retriever.retrieve(query)
        vector_nodes = self.vector_retriever.retrieve(query)
        seen = set()
        combined = []
        for n in graph_nodes + vector_nodes:
            if n.node_id not in seen:
                seen.add(n.node_id)
                combined.append(n)
        return combined

hybrid_retriever = HybridRetriever(graph_retriever, vector_retriever)
hybrid_engine = RetrieverQueryEngine.from_args(
    hybrid_retriever,
    llm=index.llm,
    node_postprocessors=[SimilarityPostprocessor(similarity_cutoff=0.6)],
)

Test the hybrid engine against the same questions plus one that benefits from vector search:

# test_hybrid.py
from hybrid_engine import hybrid_engine

questions = [
    "What models does n4n.ai route to?",
    "How does LlamaIndex extract knowledge graph triplets?",
    "What routing directives does n4n.ai support?",
    "Summarize the key features of the n4n.ai gateway.",
]

for q in questions:
    print(f"\nQ: {q}")
    response = hybrid_engine.query(q)
    print(f"A: {response.response}")
    print(f"Sources: {len(response.source_nodes)} nodes")

The last question (“Summarize…”) typically produces a better answer with hybrid retrieval because it draws from full text chunks, not just extracted triples.

Step 8: Persist and reload the index

In production, you’ll build the index once and reload it for serving. The graph store (Neo4j) persists automatically. The vector index and document store need explicit persistence.

# persist.py
from build_index import index
from hybrid_engine import vector_index

PERSIST_DIR = "./storage"

# Persist vector index and docstore
index.storage_context.persist(persist_dir=PERSIST_DIR)
vector_index.storage_context.persist(persist_dir=PERSIST_DIR)
print(f"Persisted to {PERSIST_DIR}")

Reload in a separate process:

# reload.py
import os
from llama_index.core import (
    load_index_from_storage,
    StorageContext,
    KnowledgeGraphIndex,
    VectorStoreIndex,
)
from llama_index.graph_stores.neo4j import Neo4jGraphStore
from config import llm, embed_model, storage_context as base_storage_context

PERSIST_DIR = "./storage"

# Recreate graph store connection
graph_store = Neo4jGraphStore(
    username=os.getenv("NEO4J_USER"),
    password=os.getenv("NEO4J_PASSWORD"),
    url=os.getenv("NEO4J_URI"),
    database="neo4j",
)

# Load KG index from graph store (triplets live in Neo4j)
kg_index = KnowledgeGraphIndex(
    storage_context=StorageContext.from_defaults(graph_store=graph_store),
    llm=llm,
    embed_model=embed_model,
)

# Load vector index from disk
vector_storage_context = StorageContext.from_defaults(
    persist_dir=PERSIST_DIR,
    graph_store=graph_store,
)
vector_index = load_index_from_storage(vector_storage_context, embed_model=embed_model)

print(f"KG triplets: {len(kg_index.index_struct.triplets)}")
print(f"Vector index docs: {len(vector_index.docstore.docs)}")

Run persist.py then reload.py. Counts should match.

Step 9: Add routing directives for production control

n4n.ai accepts routing directives via extra headers. In LlamaIndex, pass these through the additional_kwargs on the LLM call. The simplest approach is a wrapper that injects headers per request.

# routing.py
from llama_index.llms.openai import OpenAI
from llama_index.core.llms import CompletionResponse
from typing import Any, Dict, Optional
import os

class RoutedLLM(OpenAI):
    def __init__(self, *args, default_headers: Optional[Dict[str, str]] = None, **kwargs):
        super().__init__(*args, **kwargs)
        self.default_headers = default_headers or {}

    def complete(self, prompt: str, **kwargs) -> CompletionResponse:
        # Merge per-call headers with defaults
        extra_headers = {**self.default_headers, **kwargs.pop("extra_headers", {})}
        if extra_headers:
            kwargs["extra_headers"] = extra_headers
        return super().complete(prompt, **kwargs)

    async def acomplete(self, prompt: str, **kwargs) -> CompletionResponse:
        extra_headers = {**self.default_headers, **kwargs.pop("extra_headers", {})}
        if extra_headers:
            kwargs["extra_headers"] = extra_headers
        return await super().acomplete(prompt, **kwargs)

# Usage: prefer low latency for interactive queries
routed_llm = RoutedLLM(
    model="meta-llama/llama-3.1-70b-instruct",
    api_base=os.getenv("N4N_BASE_URL"),
    api_key=os.getenv("N4N_API_KEY"),
    temperature=0.0,
    default_headers={"x-n4n-max-latency-ms": "800"},
)

# For batch/background jobs, prefer cost
batch_llm = RoutedLLM(
    model="meta-llama/llama-3.1-70b-instruct",
    api_base=os.getenv("N4N_BASE_URL"),
    api_key=os.getenv("N4N_API_KEY"),
    temperature=0.0,
    default_headers={"x-n4n-prefer": "cost"},
)

Pass routed_llm to your query engine instead of the base llm. The gateway will honor the latency bound and route to the fastest available provider.

Step 10: Verify end-to-end with a smoke test

Create a single script that exercises the full pipeline: build (or reload), query, and assert expected behavior.

# smoke_test.py
import os
from config import llm, embed_model
from graph_store import graph_store
from documents import docs

def test_full_pipeline():
    # 1. Ensure clean graph
    graph_store.query("MATCH (n) DETACH DELETE n")
    
    # 2. Build index
    from llama_index.core import KnowledgeGraphIndex, StorageContext
    storage_context = StorageContext.from_defaults(graph_store=graph_store)
    index = KnowledgeGraphIndex.from_documents(
        docs,
        storage_context=storage_context,
        max_triplets_per_chunk=10,
        include_embeddings=True,
        show_progress=False,
    )
    assert len(index.index_struct.triplets) > 0, "No triplets extracted"
    print(f"✓ Built index with {len(index.index_struct.triplets)} triplets")
    
    # 3. Query engine
    from llama_index.core.query_engine import KnowledgeGraphQueryEngine
    qe = KnowledgeGraphQueryEngine(
        storage_context=storage_context,
        llm=llm,
        verbose=False,
    )
    
    # 4. Test queries
    test_cases = [
        ("What does n4n.ai do?", ["240+", "models", "endpoint"]),
        ("How are triplets extracted?", ["LLM", "entities", "relationships"]),
        ("What routing directives exist?", ["provider", "max_latency"]),
    ]
    
    for question, keywords in test_cases:
        resp = qe.query(question)
        answer = resp.response.lower()
        for kw in keywords:
            assert kw.lower() in answer, f"Keyword '{kw}' missing in answer: {answer}"
        print(f"✓ '{question}' -> contains {keywords}")
    
    print("\nAll smoke tests passed.")

if __name__ == "__main__":
    test_full_pipeline()

Run it:

python smoke_test.py

Output should show all checks passing. If any assertion fails, inspect the Cypher in resp.metadata.get("cypher_query") and the raw graph state in Neo4j Browser.

Step 11: Production considerations

Schema constraints

The default extraction schema is open-ended. For production, define a strict schema to reduce noise and improve query reliability:

from llama_index.core import KnowledgeGraphIndex
from llama_index.core.schema import Triple

# Define allowed entity and relation types
kg_index = KnowledgeGraphIndex.from_documents(
    docs,
    storage_context=storage_context,
    kg_triple_extract_template=CUSTOM_TEMPLATE,  # your prompt with type constraints
    max_triplets_per_chunk=15,
)

Incremental updates

Don’t rebuild the full graph on every document change. Use index.insert(document) for new docs and index.delete_ref_doc(doc_id) for removals. Both update Neo4j and the vector index atomically.

Monitoring

n4n.ai returns provider metadata in response headers (x-n4n-provider, x-n4n-latency-ms, x-n4n-cache-hit). Log these alongside your query latency to detect routing issues:

import logging
logging.getLogger("httpx").setLevel(logging.DEBUG)
# Or intercept at the OpenAI client level for structured logs

Cost control

Set max_tokens on the LLM and similarity_top_k on retrievers. The hybrid retriever in Step 7 fetches from both stores — cap each at 4-6 nodes to bound context window usage.

Verification checklist

Before considering the pipeline production-ready, confirm:

• Neo4j contains expected entities and relationships (spot-check 20+ nodes in Browser)
• KnowledgeGraphQueryEngine returns grounded answers with valid Cypher
• Hybrid engine improves recall on summary-style questions
• Persist/reload cycle preserves both graph and vector indices
• Routing directives affect provider selection (check x-n4n-provider header)
• Smoke test passes in a clean environment
• Token usage per query stays within budget (monitor via n4n.ai dashboard)

Next steps

• Swap Neo4j for FalkorDB or Kuzu if you need embedded/OLAP graph stores
• Add a CypherQueryEngine for hand-written analytical queries
• Implement entity resolution to merge duplicate nodes across documents
• Add evaluation harness (e.g., llama-index-evaluation with ground-truth QA pairs)

You now have a complete, verifiable llamaindex knowledge graph query n4n.ai tutorial pipeline that extracts, stores, and queries structured knowledge with production-grade model routing.