LlamaIndex OpenAILike class with n4n.ai models

This tutorial walks through integrating LlamaIndex’s OpenAILike class with n4n.ai’s OpenAI-compatible endpoint. You’ll build a working chat client, add streaming, wire up embeddings, and assemble a minimal RAG pipeline — all against a single base URL that routes to 240+ models with automatic fallback.

Prerequisites

• Python 3.10+
• An n4n.ai API key (get one at n4n.ai)
• Packages: llama-index, llama-index-llms-openai-like, llama-index-embeddings-openai-like

pip install llama-index llama-index-llms-openai-like llama-index-embeddings-openai-like

Set your API key as an environment variable:

export N4N_API_KEY="your-key-here"

The n4n.ai endpoint is https://api.n4n.ai/v1 — standard OpenAI format, so OpenAILike works without custom adapters.

Basic chat completion

Create chat_basic.py:

import os
from llama_index.llms.openai_like import OpenAILike

llm = OpenAILike(
    model="meta-llama/llama-3.1-70b-instruct",
    api_key=os.getenv("N4N_API_KEY"),
    api_base="https://api.n4n.ai/v1",
    is_chat_model=True,
    temperature=0.2,
    max_tokens=512,
)

response = llm.complete("Explain the difference between a coroutine and a generator in Python.")
print(response.text)

Run it:

python chat_basic.py

Expected output (truncated):

A coroutine is a function that can pause execution (via `await`) and resume later,
allowing cooperative multitasking. A generator yields values lazily via `yield`
and maintains its own stack frame. Key differences:
- Coroutines: `async def`, `await`, event-loop driven, single-threaded concurrency
- Generators: `def` with `yield`, iterator protocol, pull-based iteration
...

The model parameter accepts any model ID available on n4n.ai — provider-prefixed IDs like meta-llama/llama-3.1-70b-instruct, anthropic/claude-3.5-sonnet, google/gemini-1.5-pro all work through the same endpoint.

Streaming responses

For interactive UIs, stream tokens as they arrive. Update chat_basic.py:

import os
from llama_index.llms.openai_like import OpenAILike

llm = OpenAILike(
    model="meta-llama/llama-3.1-70b-instruct",
    api_key=os.getenv("N4N_API_KEY"),
    api_base="https://api.n4n.ai/v1",
    is_chat_model=True,
    temperature=0.2,
    max_tokens=512,
)

stream = llm.stream_complete("Write a 50-line Python script that demonstrates asyncio.gather with error handling.")
for chunk in stream:
    print(chunk.delta, end="", flush=True)
print()

Run it — you’ll see tokens print incrementally. The stream_complete method returns a generator yielding CompletionResponse objects with delta containing the new token.

Chat messages with system prompts

LlamaIndex’s ChatMessage abstraction works natively. Create chat_messages.py:

import os
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.llms import ChatMessage, MessageRole

llm = OpenAILike(
    model="anthropic/claude-3.5-sonnet",
    api_key=os.getenv("N4N_API_KEY"),
    api_base="https://api.n4n.ai/v1",
    is_chat_model=True,
    temperature=0.1,
)

messages = [
    ChatMessage(role=MessageRole.SYSTEM, content="You are a senior Python engineer. Be concise."),
    ChatMessage(role=MessageRole.USER, content="How do I fix a memory leak in a long-running asyncio application?"),
]

response = llm.chat(messages)
print(response.message.content)

Output:

Common causes and fixes:
1. **Unclosed resources** — Use `async with` for clients, connections, files. Implement `__aenter__`/`__aexit__`.
2. **Task accumulation** — Track tasks in a set; `task.add_done_callback(tasks.discard)`.
3. **Event loop references** — Avoid storing loop references in long-lived objects.
4. **Circular references** — `gc.collect()` helps; use `weakref` for caches.
5. **Third-party libs** — Profile with `tracemalloc` or `objgraph` to identify leak sources.

Embeddings

The same pattern applies to embeddings via OpenAILikeEmbedding. Create embeddings.py:

import os
import numpy as np
from llama_index.embeddings.openai_like import OpenAILikeEmbedding

embed_model = OpenAILikeEmbedding(
    model="nomic-ai/nomic-embed-text-v1.5",
    api_key=os.getenv("N4N_API_KEY"),
    api_base="https://api.n4n.ai/v1",
)

texts = [
    "LlamaIndex provides data connectors for LLMs.",
    "n4n.ai routes requests across 240+ models.",
    "Python's asyncio enables concurrent I/O.",
]

embeddings = embed_model.get_text_embedding_batch(texts)
print(f"Shape: {np.array(embeddings).shape}")
print(f"First vector (first 5 dims): {embeddings[0][:5]}")

Run it:

Shape: (3, 768)
First vector (first 5 dims): [-0.0123, 0.0456, -0.0078, 0.0234, -0.0091]

The nomic-ai/nomic-embed-text-v1.5 model outputs 768-dimensional vectors. Batch embedding is more efficient than looping get_text_embedding.

Minimal RAG pipeline

Now assemble a complete retrieval-augmented generation pipeline. Create rag_pipeline.py:

import os
from pathlib import Path

from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    StorageContext,
    Settings,
)
from llama_index.llms.openai_like import OpenAILike
from llama_index.embeddings.openai_like import OpenAILikeEmbedding

# Configure global settings
Settings.llm = OpenAILike(
    model="meta-llama/llama-3.1-70b-instruct",
    api_key=os.getenv("N4N_API_KEY"),
    api_base="https://api.n4n.ai/v1",
    is_chat_model=True,
    temperature=0.1,
    max_tokens=1024,
)

Settings.embed_model = OpenAILikeEmbedding(
    model="nomic-ai/nomic-embed-text-v1.5",
    api_key=os.getenv("N4N_API_KEY"),
    api_base="https://api.n4n.ai/v1",
)

# Load documents
documents = SimpleDirectoryReader("./data").load_data()
print(f"Loaded {len(documents)} documents")

# Build index
index = VectorStoreIndex.from_documents(documents)
print("Index built")

# Persist for reuse
index.storage_context.persist(persist_dir="./storage")
print("Index persisted to ./storage")

# Query
query_engine = index.as_query_engine(similarity_top_k=3)
response = query_engine.query("What are the key components of LlamaIndex?")
print(f"\nAnswer: {response.response}")
print(f"\nSources: {[node.metadata.get('file_name') for node in response.source_nodes]}")

Create a data/ directory with some .txt or .md files, then run:

mkdir -p data
echo "LlamaIndex core components: Data Connectors (ingest), Indexes (structure), Engines (query), Agents (act), Observability (monitor)." > data/overview.txt
echo "VectorStoreIndex stores embeddings in a vector database for semantic search." > data/indexes.txt
python rag_pipeline.py

Output:

Loaded 2 documents
Index built
Index persisted to ./storage

Answer: LlamaIndex's key components are Data Connectors for ingestion, Indexes for structuring data, Engines for querying, Agents for automated reasoning, and Observability for monitoring.

Sources: ['overview.txt', 'indexes.txt']

On subsequent runs, load the persisted index instead of rebuilding:

from llama_index.core import load_index_from_storage

storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)

Model routing and fallback

n4n.ai honors client routing directives. You can specify a provider preference via the model parameter or let the gateway handle automatic fallback when a provider is rate-limited or degraded. For explicit routing, use provider-prefixed model IDs:

# Prefer Anthropic, fallback handled by gateway
llm = OpenAILike(
    model="anthropic/claude-3.5-sonnet",
    api_key=os.getenv("N4N_API_KEY"),
    api_base="https://api.n4n.ai/v1",
    is_chat_model=True,
)

# Or let the gateway choose the best available
llm = OpenAILike(
    model="meta-llama/llama-3.1-70b-instruct",  # primary
    api_key=os.getenv("N4N_API_KEY"),
    api_base="https://api.n4n.ai/v1",
    is_chat_model=True,
)

The gateway forwards provider cache-control hints, so repeated prompts with identical prefixes can hit KV caches on supported models (Anthropic, Google) without client-side changes.

Error handling and retries

Wrap calls for production resilience. Create robust_client.py:

import os
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.llms import ChatMessage, MessageRole

llm = OpenAILike(
    model="meta-llama/llama-3.1-70b-instruct",
    api_key=os.getenv("N4N_API_KEY"),
    api_base="https://api.n4n.ai/v1",
    is_chat_model=True,
    temperature=0.2,
    max_tokens=512,
    timeout=60.0,
)

@retry(
    wait=wait_exponential_jitter(initial=1, max=30),
    stop=stop_after_attempt(3),
)
async def safe_chat(messages: list[ChatMessage]) -> str:
    response = await llm.achat(messages)
    return response.message.content

async def main():
    messages = [
        ChatMessage(role=MessageRole.SYSTEM, content="Answer in one sentence."),
        ChatMessage(role=MessageRole.USER, content="What is the capital of France?"),
    ]
    try:
        answer = await safe_chat(messages)
        print(answer)
    except Exception as e:
        print(f"Failed after retries: {e}")

asyncio.run(main())

Add tenacity to requirements:

pip install tenacity

The async achat/astream_chat methods integrate cleanly with FastAPI, Starlette, or any async framework.

Production checklist

Before deploying:

Next steps

• Swap VectorStoreIndex for a managed vector DB (Pinecone, Weaviate, Qdrant) via LlamaIndex integrations.
• Add QueryPipeline for multi-step reasoning (rewrite → retrieve → rerank → synthesize).
• Implement streaming RAG with astream_chat and a React/Vue frontend.
• Explore FunctionCallingAgent with tools for code execution, web search, or database queries.

The OpenAILike class keeps your LlamaIndex code portable — swap the api_base and model to point at any OpenAI-compatible endpoint without rewriting your application logic.