LlamaIndex query engine with Claude 3.5 Sonnet on n4n.ai

This tutorial walks through building a retrieval-augmented generation pipeline with LlamaIndex’s query engine abstraction, using Claude 3.5 Sonnet as the LLM and n4n.ai as the inference gateway. You’ll end up with a modular, observable RAG system that handles document ingestion, vector indexing, and multi-step query planning — all behind a single OpenAI-compatible endpoint that routes across 240+ models with automatic fallback.

Prerequisites

• Python 3.10+
• An n4n.ai API key (get one at n4n.ai)
• Basic familiarity with LlamaIndex concepts: documents, nodes, indexes, query engines

Install dependencies:

pip install llama-index llama-index-llms-openai llama-index-embeddings-openai \
    llama-index-vector-stores-chroma pypdf python-dotenv

We use the OpenAI-compatible packages because n4n.ai exposes an OpenAI-compatible endpoint. No custom LlamaIndex integrations required.

Project structure

rag-cli/
├── .env
├── config.py
├── ingest.py
├── query.py
└── data/
    └── (your PDFs here)

Configuration

Create .env:

N4N_API_KEY=your_key_here
N4N_BASE_URL=https://api.n4n.ai/v1
EMBED_MODEL=text-embedding-3-large
LLM_MODEL=anthropic/claude-3.5-sonnet
CHROMA_PATH=./chroma_db

config.py centralizes settings and builds the LlamaIndex components:

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

load_dotenv()

N4N_API_KEY = os.getenv("N4N_API_KEY")
N4N_BASE_URL = os.getenv("N4N_BASE_URL", "https://api.n4n.ai/v1")
EMBED_MODEL = os.getenv("EMBED_MODEL", "text-embedding-3-large")
LLM_MODEL = os.getenv("LLM_MODEL", "anthropic/claude-3.5-sonnet")
CHROMA_PATH = os.getenv("CHROMA_PATH", "./chroma_db")

def configure_settings() -> None:
    """Configure global LlamaIndex settings to use n4n.ai."""
    Settings.llm = OpenAI(
        model=LLM_MODEL,
        api_key=N4N_API_KEY,
        api_base=N4N_BASE_URL,
        temperature=0.1,
        max_tokens=4096,
    )
    Settings.embed_model = OpenAIEmbedding(
        model=EMBED_MODEL,
        api_key=N4N_API_KEY,
        api_base=N4N_BASE_URL,
    )
    Settings.chunk_size = 1024
    Settings.chunk_overlap = 128

The key insight: OpenAI from llama_index.llms.openai works unchanged because n4n.ai honors the OpenAI chat completions contract. The model string anthropic/claude-3.5-sonnet is a routing directive — n4n.ai resolves it to the appropriate provider and handles fallback if that provider is degraded.

Document ingestion

ingest.py loads PDFs, splits them into nodes, and persists a Chroma vector store:

# ingest.py
import os
from pathlib import Path
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    StorageContext,
)
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

from config import CHROMA_PATH, configure_settings

configure_settings()

DATA_DIR = Path("./data")
CHROMA_COLLECTION = "rag_docs"

def build_index() -> VectorStoreIndex:
    """Load documents, create nodes, persist to Chroma."""
    if not DATA_DIR.exists():
        raise FileNotFoundError(f"Data directory {DATA_DIR} does not exist")

    documents = SimpleDirectoryReader(
        input_dir=str(DATA_DIR),
        required_exts=[".pdf"],
        recursive=True,
    ).load_data()

    print(f"Loaded {len(documents)} documents")

    # Initialize Chroma client and collection
    chroma_client = chromadb.PersistentClient(path=CHROMA_PATH)
    chroma_collection = chroma_client.get_or_create_collection(CHROMA_COLLECTION)
    vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
    storage_context = StorageContext.from_defaults(vector_store=vector_store)

    index = VectorStoreIndex.from_documents(
        documents,
        storage_context=storage_context,
        show_progress=True,
    )

    print(f"Index built with {len(index.docstore.docs)} nodes")
    return index

if __name__ == "__main__":
    build_index()

Run it:

python ingest.py

Expected output:

Loaded 7 documents
Index built with 247 nodes

The node count depends on your PDFs and the chunk settings in config.py. Each node is a 1024-token chunk with 128-token overlap.

Query engine with sub-question planning

LlamaIndex’s SubQuestionQueryEngine decomposes complex queries into sub-questions, executes each against the vector index, and synthesizes a final answer. This is where Claude 3.5 Sonnet’s reasoning shines.

query.py:

# query.py
import sys
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.query_engine import SubQuestionQueryEngine
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

from config import CHROMA_PATH, CHROMA_COLLECTION, configure_settings

configure_settings()

def load_index() -> VectorStoreIndex:
    """Load persisted index from Chroma."""
    chroma_client = chromadb.PersistentClient(path=CHROMA_PATH)
    chroma_collection = chroma_client.get_collection(CHROMA_COLLECTION)
    vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
    storage_context = StorageContext.from_defaults(vector_store=vector_store)
    return VectorStoreIndex.from_vector_store(vector_store, storage_context=storage_context)

def build_query_engine(index: VectorStoreIndex) -> SubQuestionQueryEngine:
    """Construct a sub-question query engine over the index."""
    base_query_engine = index.as_query_engine(
        similarity_top_k=8,
        response_mode="compact",
    )

    query_engine_tool = QueryEngineTool(
        query_engine=base_query_engine,
        metadata=ToolMetadata(
            name="document_store",
            description=(
                "Useful for answering questions about the ingested PDF documents. "
                "Contains technical specifications, API references, and architectural decisions."
            ),
        ),
    )

    return SubQuestionQueryEngine.from_defaults(
        query_engine_tools=[query_engine_tool],
        use_async=True,
        verbose=True,
    )

def main():
    if len(sys.argv) < 2:
        print("Usage: python query.py \"your question here\"")
        sys.exit(1)

    question = " ".join(sys.argv[1:])
    print(f"\nQuestion: {question}\n")

    index = load_index()
    query_engine = build_query_engine(index)

    response = query_engine.query(question)

    print("\n" + "=" * 60)
    print("FINAL ANSWER")
    print("=" * 60)
    print(response.response)
    print("\n" + "=" * 60)
    print("SOURCE NODES")
    print("=" * 60)
    for i, node in enumerate(response.source_nodes):
        print(f"\n--- Source {i+1} (score: {node.score:.3f}) ---")
        print(node.text[:300] + "..." if len(node.text) > 300 else node.text)

if __name__ == "__main__":
    main()

Run a test query:

python query.py "How does the authentication flow work in the API v2 spec?"

Expected output (truncated for brevity):

Question: How does the authentication flow work in the API v2 spec?

=== Sub Question 1 ===
What authentication methods are supported in API v2?
=== Sub Question 2 ===
How do I obtain and refresh access tokens?
=== Sub Question 3 ===
What are the required scopes for each endpoint?

FINAL ANSWER
============================================================
API v2 uses OAuth 2.0 with PKCE for public clients and client credentials
for service-to-service communication. The flow:

1. **Authorization Code Grant with PKCE** (recommended for SPAs/mobile):
   - Generate code_verifier and code_challenge (S256)
   - Redirect user to `/oauth/authorize` with challenge
   - Exchange authorization code for access + refresh token at `/oauth/token`

2. **Client Credentials Grant** (machine-to-machine):
   - POST to `/oauth/token` with client_id, client_secret, grant_type=client_credentials
   - Returns access token only (no refresh token)

Access tokens are JWTs with 15-minute TTL. Refresh tokens rotate on use and
expire after 30 days of inactivity. Required scopes are documented per-endpoint
in the OpenAPI spec — `read:resources` for GET, `write:resources` for mutations.

SOURCE NODES
============================================================

--- Source 1 (score: 0.892) ---
Authentication in API v2 follows OAuth 2.0 RFC 6749 with PKCE extension
(RFC 7636). The authorization server issues JWT access tokens signed with
RS256. Public clients MUST use PKCE...

The sub-question decomposition is visible because verbose=True. Each sub-question hits the vector index independently, then Claude synthesizes the final answer. This beats a single-shot retrieval for multi-faceted questions.

Streaming responses

For CLI or web UIs, stream tokens as they arrive:

# query_streaming.py (add to query.py or separate file)
def main_streaming():
    # ... same setup ...
    response = query_engine.query(question)
    
    print("\nStreaming response:")
    for token in response.response_gen:
        print(token, end="", flush=True)
    print()

The response_gen attribute exists on SubQuestionQueryEngine responses when the underlying LLM supports streaming (Claude 3.5 Sonnet does via n4n.ai).

Observability: what n4n.ai surfaces

Since n4n.ai sits between your code and the model providers, you get provider-agnostic observability without instrumentation code. Each response includes headers:

• x-n4n-model: the actual model that served the request (e.g., anthropic/claude-3.5-sonnet-20241022)
• x-n4n-provider: the upstream provider (anthropic, aws-bedrock, etc.)
• x-n4n-tokens-prompt / x-n4n-tokens-completion: token counts for cost tracking
• x-n4n-latency-ms: end-to-end latency

LlamaIndex’s OpenAI client captures these automatically if you enable response metadata:

from llama_index.core.callbacks import CallbackManager, TokenCountingHandler

token_counter = TokenCountingHandler()
Settings.callback_manager = CallbackManager([token_counter])

# After a query:
print(f"Prompt tokens: {token_counter.prompt_llm_token_count}")
print(f"Completion tokens: {token_counter.completion_llm_token_count}")

For production, consider wiring n4n.ai’s usage headers into your own metrics pipeline — they’re more accurate than client-side estimates because they reflect what the provider actually billed.

Common pitfalls

Chunk size too small for Claude’s context window.
Claude 3.5 Sonnet has a 200k context window. The default 1024-token chunks are conservative. For dense technical PDFs, try chunk_size=2048 with chunk_overlap=256 to reduce fragmentation.

Embedding model mismatch.
If you switch embedding models (e.g., text-embedding-3-small for cost), you must re-ingest. Chroma doesn’t support multiple embedding spaces in one collection.

Rate limits on the embedding endpoint.
n4n.ai falls back across providers for chat completions, but embedding calls go to a single provider per model. If you hit 429s during ingestion, add a retry wrapper or batch your SimpleDirectoryReader calls.

Sub-question engine loops.
Complex queries can generate many sub-questions. Set a hard limit:

SubQuestionQueryEngine.from_defaults(
    query_engine_tools=[query_engine_tool],
    use_async=True,
    verbose=True,
    max_sub_questions=5,  # prevent runaway decomposition
)

Extending: hybrid search

Vector similarity alone misses exact matches (error codes, function names). Add a BM25 retriever and fuse results:

from llama_index.core.retrievers import VectorIndexRetriever, BM25Retriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import SimilarityPostprocessor

def build_hybrid_query_engine(index: VectorStoreIndex):
    vector_retriever = VectorIndexRetriever(index=index, similarity_top_k=8)
    bm25_retriever = BM25Retriever.from_defaults(
        docstore=index.docstore,
        similarity_top_k=8,
    )

    # Simple fusion: combine node sets, deduplicate by node_id
    class HybridRetriever:
        def __init__(self, *retrievers):
            self.retrievers = retrievers

        def retrieve(self, query_bundle):
            all_nodes = []
            seen = set()
            for r in self.retrievers:
                for node in r.retrieve(query_bundle):
                    if node.node_id not in seen:
                        seen.add(node.node_id)
                        all_nodes.append(node)
            return all_nodes[:12]  # cap total

    hybrid = HybridRetriever(vector_retriever, bm25_retriever)
    return RetrieverQueryEngine.from_args(
        hybrid,
        node_postprocessors=[SimilarityPostprocessor(similarity_cutoff=0.5)],
    )

Swap base_query_engine in build_query_engine() with the hybrid version. The sub-question planner works unchanged.

Production checklist

• Persist Chroma to a managed volume (not local disk in containers)
• Add structured logging for queries, latency, and token counts
• Implement query validation: reject empty or overly long inputs
• Set up n4n.ai routing directives per workload (e.g., provider: anthropic for reasoning, provider: aws-bedrock for compliance)
• Monitor x-n4n-provider header to detect fallback events
• Schedule periodic re-ingestion for living documents

Summary

You now have a LlamaIndex RAG pipeline that:

1. Ingests PDFs into a persistent Chroma vector store using n4n.ai-hosted embeddings
2. Exposes a SubQuestionQueryEngine that decomposes complex queries and synthesizes answers with Claude 3.5 Sonnet
3. Streams responses token-by-token for interactive UIs
4. Surfaces provider-level observability via n4n.ai response headers

The entire LLM and embedding stack routes through one OpenAI-compatible endpoint. Swap models by changing LLM_MODEL in .env — no code changes.