n4nAI
Open console →
Framework tutorialsTutorialLlamaIndex Chat Engines & Memory

LlamaIndex chat engine with n4n.ai and Claude models

Build a production-ready LlamaIndex chat engine using n4n.ai's OpenAI-compatible endpoint to access Claude models with automatic fallback and token metering.

n4n Team3 min read693 words

Audio narration

Coming soon — every post will get a voice note here.

If you’re building a conversational agent with LlamaIndex, you’ve probably hit the wall where OpenAI’s API becomes a single point of failure — rate limits, regional outages, or pricing that doesn’t match your traffic profile. n4n.ai solves this by exposing 240+ models behind one OpenAI-compatible endpoint, including Anthropic’s Claude family, with automatic fallback when a provider degrades. This tutorial walks through wiring a LlamaIndex chat engine to n4n.ai, adding conversation memory, and handling streaming responses correctly.

Prerequisites

You need Python 3.10+ and an n4n.ai API key. Get one at n4n.ai if you don’t have it. Install the dependencies:

pip install llama-index llama-index-llms-openai python-dotenv

Create a .env file in your project root:

N4N_API_KEY=your_n4n_api_key_here
N4N_BASE_URL=https://api.n4n.ai/v1

The base URL is the only configuration change from a standard OpenAI setup — everything else uses LlamaIndex’s native OpenAI integration.

Minimal working example

Start with a script that proves the connection works. Save this as chat_basic.py:

import os
from dotenv import load_dotenv
from llama_index.llms.openai import OpenAI
from llama_index.core.chat_engine import SimpleChatEngine
from llama_index.core.memory import ChatMemoryBuffer

load_dotenv()

llm = OpenAI(
    model="claude-3-5-sonnet-20241022",
    api_key=os.getenv("N4N_API_KEY"),
    api_base=os.getenv("N4N_BASE_URL"),
    temperature=0.3,
)

memory = ChatMemoryBuffer.from_defaults(token_limit=3000)
chat_engine = SimpleChatEngine.from_defaults(llm=llm, memory=memory)

response = chat_engine.chat("What is the capital of France?")
print(response.response)

Run it:

python chat_basic.py

Expected output:

The capital of France is Paris.

The ChatMemoryBuffer with a 3000-token limit keeps the last several turns in context. Adjust token_limit based on your model’s context window and cost tolerance — Claude 3.5 Sonnet supports 200K tokens, but you rarely need that much for chat history.

Streaming responses

Blocking on the full response kills perceived latency. LlamaIndex’s stream_chat yields tokens as they arrive. Update chat_basic.py:

import os
from dotenv import load_dotenv
from llama_index.llms.openai import OpenAI
from llama_index.core.chat_engine import SimpleChatEngine
from llama_index.core.memory import ChatMemoryBuffer

load_dotenv()

llm = OpenAI(
    model="claude-3-5-sonnet-20241022",
    api_key=os.getenv("N4N_API_KEY"),
    api_base=os.getenv("N4N_BASE_URL"),
    temperature=0.3,
    streaming=True,
)

memory = ChatMemoryBuffer.from_defaults(token_limit=3000)
chat_engine = SimpleChatEngine.from_defaults(llm=llm, memory=memory)

print("Chat started. Type 'exit' to quit.\n")
while True:
    user_input = input("You: ")
    if user_input.lower() in ("exit", "quit"):
        break

    print("Assistant: ", end="", flush=True)
    streaming_response = chat_engine.stream_chat(user_input)
    for token in streaming_response.response_gen:
        print(token, end="", flush=True)
    print("\n")

Run it and you’ll see tokens appear character-by-character. The streaming=True flag on the OpenAI constructor is the only change required — LlamaIndex handles the SSE parsing internally.

Adding a system prompt

Production chat engines need a system prompt to define behavior, tone, and guardrails. Pass it at engine creation:

from llama_index.core.chat_engine import SimpleChatEngine
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.llms import ChatMessage, MessageRole

system_prompt = """You are a senior Python engineer. Answer concisely.
Prefer code examples over explanations. Never hallucinate imports."""

memory = ChatMemoryBuffer.from_defaults(
    token_limit=3000,
    chat_history=[
        ChatMessage(role=MessageRole.SYSTEM, content=system_prompt)
    ],
)

chat_engine = SimpleChatEngine.from_defaults(llm=llm, memory=memory)

The system message sits at index 0 of chat_history and persists across turns. ChatMemoryBuffer evicts oldest user/assistant pairs first, never the system message.

Using the condense-plus-context chat engine

SimpleChatEngine stuffs the entire conversation into the prompt. For longer sessions, switch to CondensePlusContextChatEngine — it summarizes history into a standalone question, retrieves relevant context (if you attach an index), then answers. Here’s the setup without a vector index (pure chat):

from llama_index.core.chat_engine import CondensePlusContextChatEngine
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.llms import ChatMessage, MessageRole

memory = ChatMemoryBuffer.from_defaults(token_limit=3000)

chat_engine = CondensePlusContextChatEngine.from_defaults(
    llm=llm,
    memory=memory,
    system_prompt=system_prompt,
    verbose=True,
)

verbose=True logs the condensed question and context to stdout — useful for debugging. When you later add a vector index, pass retriever=index.as_retriever() to enable RAG.

Handling provider fallback gracefully

n4n.ai returns standard OpenAI error codes. When a provider hits rate limits or degrades, n4n.ai automatically routes to a healthy backend and surfaces the result transparently. You only need to handle the usual RateLimitError and APIConnectionError from the OpenAI SDK. Wrap your chat loop:

import openai
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

@retry(
    wait=wait_exponential_jitter(initial=1, max=30),
    stop=stop_after_attempt(3),
    retry=(
        lambda e: isinstance(e, (openai.RateLimitError, openai.APIConnectionError))
    ),
)
def safe_chat(engine, message: str):
    return engine.chat(message)

# In your loop:
try:
    response = safe_chat(chat_engine, user_input)
    print(response.response)
except openai.RateLimitError:
    print("All backends rate-limited. Try again in a moment.")
except openai.APIConnectionError:
    print("Network issue reaching n4n.ai. Check connectivity.")

The tenacity retry policy backs off exponentially with jitter — standard practice for LLM gateways. Install it with pip install tenacity if not present.

Token usage and cost tracking

n4n.ai returns usage in the standard OpenAI usage field. LlamaIndex exposes it on the response object:

response = chat_engine.chat("Explain asyncio in 50 words.")
print(f"Prompt tokens: {response.raw.usage.prompt_tokens}")
print(f"Completion tokens: {response.raw.usage.completion_tokens}")
print(f"Total tokens: {response.raw.usage.total_tokens}")

Sample output:

Prompt tokens: 42
Completion tokens: 67
Total tokens: 109

For streaming, usage arrives in the final chunk. Access it after the loop:

streaming_response = chat_engine.stream_chat(user_input)
full_response = ""
for token in streaming_response.response_gen:
    print(token, end="", flush=True)
    full_response += token
print("\n")

usage = streaming_response.raw.usage
print(f"Total tokens this turn: {usage.total_tokens}")

Persisting memory across restarts

ChatMemoryBuffer is in-memory. For production, serialize to disk or a database. LlamaIndex provides ChatMemoryBuffer.to_dict() and from_dict():

import json
from pathlib import Path

MEMORY_PATH = Path("chat_memory.json")

def load_memory():
    if MEMORY_PATH.exists():
        data = json.loads(MEMORY_PATH.read_text())
        return ChatMemoryBuffer.from_dict(data)
    return ChatMemoryBuffer.from_defaults(token_limit=3000)

def save_memory(memory: ChatMemoryBuffer):
    MEMORY_PATH.write_text(json.dumps(memory.to_dict()))

memory = load_memory()
# ... run chat loop ...
save_memory(memory)

This survives process restarts. For multi-user deployments, swap the file for Redis or Postgres — the dict schema is the same.

Structured output with Pydantic

If you need guaranteed JSON from the model, use LlamaIndex’s structured_predict with a Pydantic model. This works through n4n.ai because Claude supports tool calling:

from pydantic import BaseModel, Field
from llama_index.core.program import LLMTextCompletionProgram

class CodeReview(BaseModel):
    summary: str = Field(description="One-sentence summary")
    issues: list[str] = Field(description="List of concrete issues")
    severity: str = Field(description="critical, major, minor, or none")

program = LLMTextCompletionProgram.from_defaults(
    output_cls=CodeReview,
    llm=llm,
    prompt_template_str=(
        "Review this Python code for bugs and style issues:\n{code}\n\n"
        "Return structured output only."
    ),
    verbose=True,
)

code_snippet = """
def fetch_users(db):
    return db.execute("SELECT * FROM users WHERE active = 1")
"""

result = program(code=code_snippet)
print(result.summary)
print(result.issues)
print(result.severity)

Output:

SQL injection risk via string interpolation; missing type hints.
['Use parameterized queries', 'Add type annotations for db and return type']
major

Running behind a reverse proxy

If your infrastructure requires a corporate proxy, configure the OpenAI client’s http_client:

import httpx

proxy_client = httpx.Client(proxy="http://corporate-proxy:3128")

llm = OpenAI(
    model="claude-3-5-sonnet-20241022",
    api_key=os.getenv("N4N_API_KEY"),
    api_base=os.getenv("N4N_BASE_URL"),
    http_client=proxy_client,
    streaming=True,
)

The http_client parameter accepts any httpx.Client or httpx.AsyncClient instance — useful for custom TLS, timeouts, or retry middleware.

Switching models at runtime

n4n.ai’s model routing lets you change models without code changes. Expose the model name via environment variable:

import os

MODEL = os.getenv("N4N_MODEL", "claude-3-5-sonnet-20241022")

llm = OpenAI(
    model=MODEL,
    api_key=os.getenv("N4N_API_KEY"),
    api_base=os.getenv("N4N_BASE_URL"),
    temperature=0.3,
    streaming=True,
)

Deploy the same container with N4N_MODEL=claude-3-haiku-20240307 for lower latency, or claude-3-opus-20240229 for complex reasoning. No redeploy needed if you use a config map or secret store.

Complete production-ready script

Here’s everything combined — streaming, memory persistence, retry logic, usage logging, and model switching:

import os
import json
from pathlib import Path
from dotenv import load_dotenv
import openai
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
from llama_index.llms.openai import OpenAI
from llama_index.core.chat_engine import CondensePlusContextChatEngine
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.llms import ChatMessage, MessageRole

load_dotenv()

N4N_API_KEY = os.getenv("N4N_API_KEY")
N4N_BASE_URL = os.getenv("N4N_BASE_URL", "https://api.n4n.ai/v1")
MODEL = os.getenv("N4N_MODEL", "claude-3-5-sonnet-20241022")
MEMORY_PATH = Path("chat_memory.json")

SYSTEM_PROMPT = """You are a senior Python engineer. Answer concisely.
Prefer code examples over explanations. Never hallucinate imports."""

llm = OpenAI(
    model=MODEL,
    api_key=N4N_API_KEY,
    api_base=N4N_BASE_URL,
    temperature=0.3,
    streaming=True,
)

def load_memory():
    if MEMORY_PATH.exists():
        data = json.loads(MEMORY_PATH.read_text())
        return ChatMemoryBuffer.from_dict(data)
    return ChatMemoryBuffer.from_defaults(
        token_limit=3000,
        chat_history=[ChatMessage(role=MessageRole.SYSTEM, content=SYSTEM_PROMPT)],
    )

def save_memory(memory: ChatMemoryBuffer):
    MEMORY_PATH.write_text(json.dumps(memory.to_dict()))

@retry(
    wait=wait_exponential_jitter(initial=1, max=30),
    stop=stop_after_attempt(3),
    retry=lambda e: isinstance(e, (openai.RateLimitError, openai.APIConnectionError)),
)
def safe_stream_chat(engine, message: str):
    return engine.stream_chat(message)

def main():
    memory = load_memory()
    chat_engine = CondensePlusContextChatEngine.from_defaults(
        llm=llm,
        memory=memory,
        system_prompt=SYSTEM_PROMPT,
        verbose=False,
    )

    print(f"Chat ready (model: {MODEL}). Type 'exit' to quit.\n")
    while True:
        try:
            user_input = input("You: ")
        except (EOFError, KeyboardInterrupt):
            break

        if user_input.lower() in ("exit", "quit"):
            break

        print("Assistant: ", end="", flush=True)
        try:
            streaming_response = safe_stream_chat(chat_engine, user_input)
            full_response = ""
            for token in streaming_response.response_gen:
                print(token, end="", flush=True)
                full_response += token
            print("\n")

            usage = streaming_response.raw.usage
            print(f"[tokens: {usage.total_tokens} | prompt: {usage.prompt_tokens} | completion: {usage.completion_tokens}]\n")

        except openai.RateLimitError:
            print("All backends rate-limited. Try again in a moment.\n")
        except openai.APIConnectionError:
            print("Network issue reaching n4n.ai. Check connectivity.\n")

    save_memory(memory)
    print("Memory saved. Goodbye.")

if __name__ == "__main__":
    main()

Run it:

python chat_production.py

Sample session:

Chat ready (model: claude-3-5-sonnet-20241022). Type 'exit' to quit.

You: Write a context manager that times a block.
Assistant: import time
from contextlib import contextmanager

@contextmanager
def timer():
    start = time.perf_counter()
    yield
    print(f"Elapsed: {time.perf_counter() - start:.3f}s")

[tokens: 87 | prompt: 42 | completion: 45]

You: exit
Memory saved. Goodbye.

What to do next

  • Attach a vector index to CondensePlusContextChatEngine via retriever=index.as_retriever() for RAG over your docs.
  • Swap ChatMemoryBuffer for VectorMemory if you want semantic history retrieval instead of sliding window.
  • Add OpenTelemetry tracing — LlamaIndex instruments spans automatically; n4n.ai forwards provider cache-control hints in response headers for cache-aware routing.
  • Load-test with your actual traffic pattern. The single-endpoint abstraction means you benchmark once, not per provider.

The pattern here — OpenAI-compatible client, standard LlamaIndex engines, retry wrapper — applies to any model n4n.ai serves. Change the model string, not the code.

Tagsllamaindexchat-enginen4n-aiclaude

Written by

n4n Team

The team building n4n — a single OpenAI-compatible API in front of 240+ models, with automatic fallback, load balancing and pay-per-token metering.

More from n4n Team →
Framework tutorialsMulti-turn RAG conversations in LlamaIndexBuild multi-turn RAG conversations in LlamaIndex with chat engines, memory buffers, and context compression — practical patterns for production systems.Framework tutorialsLlamaIndex chat engine with streaming responsesBuild a LlamaIndex chat engine with streaming responses — step-by-step setup, memory integration, and verification.Framework tutorialsLlamaIndex chat memory buffer: how it worksA practical guide to LlamaIndex's ChatMemoryBuffer — how token limits work, when to use summarization, and common pitfalls in production chat applications.Framework tutorialsLlamaIndex SimpleChatEngine vs ContextChatEngineTechnical comparison of LlamaIndex SimpleChatEngine and ContextChatEngine — capabilities, trade-offs, and when to use each.Framework tutorialsAdd long-term memory to a LlamaIndex chat engineLearn how to add persistent cross-session memory to a LlamaIndex chat engine with Redis in this hands-on llamaindex long-term memory chat engine tutorial.Framework tutorialsBuild a chatbot with LlamaIndex CondensePlusContextStep-by-step llamaindex condensepluscontext chatbot tutorial: build a context-aware chat engine with LlamaIndex, handle conversation history, and run it.

All llamaindex chat engines & memory posts →