Persisting LlamaIndex chat history across sessions

LlamaIndex chat engines are stateless by default — every request starts with a blank slate unless you explicitly wire in memory. For production applications, you need conversation history to survive process restarts, scale across instances, and persist per-user context. This tutorial walks through the complete implementation: choosing the right memory class, connecting a durable backend, and restoring history on each new session.

Step 1: Choose your memory type

LlamaIndex ships with several memory implementations. For most production workloads, ChatMemoryBuffer is the right starting point — it maintains a rolling window of recent messages and optionally summarizes older turns to stay within token limits.

from llama_index.core.memory import ChatMemoryBuffer

memory = ChatMemoryBuffer.from_defaults(
    token_limit=3000,           # hard cap on context window
    tokenizer_fn=tokenizer,     # optional: pass your model's tokenizer for accurate counting
)

If you need full history without summarization, use ChatMemoryBuffer with a high token_limit and disable summarization. For agentic workflows that need structured memory (facts, preferences, tasks), look at VectorMemory or SimpleComposableMemory — but those are separate tutorials.

Key decision: token_limit should leave headroom for the system prompt, retrieved context, and the model’s response. For a 8k context model, 3000–4000 tokens is a safe buffer.

Step 2: Set up a persistence backend

LlamaIndex memory implements BaseMemory which requires get, put, and reset — but persistence is your responsibility. Three practical backends cover most needs:

Option A: Redis (recommended for multi-instance deployments)

# requirements: redis>=5.0
import redis
import json
from typing import List
from llama_index.core.llms import ChatMessage

class RedisChatMemory(ChatMemoryBuffer):
    def __init__(self, redis_url: str, session_id: str, token_limit: int = 3000, ttl_seconds: int = 86400 * 30):
        super().__init__(token_limit=token_limit)
        self._client = redis.from_url(redis_url, decode_responses=True)
        self._key = f"chat_memory:{session_id}"
        self._ttl = ttl_seconds
        self._load()

    def _load(self):
        data = self._client.get(self._key)
        if data:
            messages = [ChatMessage(**m) for m in json.loads(data)]
            self.chat_store.store_messages(self._key, messages)

    def _persist(self):
        messages = self.chat_store.get_messages(self._key)
        serialized = [m.model_dump() for m in messages]
        self._client.setex(self._key, self._ttl, json.dumps(serialized))

    def put(self, message: ChatMessage) -> None:
        super().put(message)
        self._persist()

    def get(self, input: str = "") -> List[ChatMessage]:
        return super().get(input)

    def reset(self) -> None:
        super().reset()
        self._client.delete(self._key)

Option B: PostgreSQL (if you already run Postgres)

# requirements: psycopg2-binary, sqlalchemy
import json
from sqlalchemy import create_engine, Column, String, Text, DateTime
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime, timezone

Base = declarative_base()

class ChatHistory(Base):
    __tablename__ = "chat_history"
    session_id = Column(String(64), primary_key=True)
    messages = Column(Text, nullable=False)
    updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))

class PostgresChatMemory(ChatMemoryBuffer):
    def __init__(self, db_url: str, session_id: str, token_limit: int = 3000):
        super().__init__(token_limit=token_limit)
        self._engine = create_engine(db_url)
        self._Session = sessionmaker(bind=self._engine)
        self._session_id = session_id
        Base.metadata.create_all(self._engine)
        self._load()

    def _load(self):
        with self._Session() as session:
            row = session.query(ChatHistory).filter_by(session_id=self._session_id).first()
            if row:
                messages = [ChatMessage(**m) for m in json.loads(row.messages)]
                self.chat_store.store_messages(self._session_id, messages)

    def _persist(self):
        messages = self.chat_store.get_messages(self._session_id)
        serialized = json.dumps([m.model_dump() for m in messages])
        with self._Session() as session:
            row = session.query(ChatHistory).filter_by(session_id=self._session_id).first()
            if row:
                row.messages = serialized
                row.updated_at = datetime.now(timezone.utc)
            else:
                row = ChatHistory(session_id=self._session_id, messages=serialized)
                session.add(row)
            session.commit()

    def put(self, message: ChatMessage) -> None:
        super().put(message)
        self._persist()

    def get(self, input: str = "") -> List[ChatMessage]:
        return super().get(input)

    def reset(self) -> None:
        super().reset()
        with self._Session() as session:
            session.query(ChatHistory).filter_by(session_id=self._session_id).delete()
            session.commit()

Option C: Local file (dev / single-instance only)

import json
from pathlib import Path
from llama_index.core.llms import ChatMessage

class FileChatMemory(ChatMemoryBuffer):
    def __init__(self, path: Path, session_id: str, token_limit: int = 3000):
        super().__init__(token_limit=token_limit)
        self._file = path / f"{session_id}.json"
        self._file.parent.mkdir(parents=True, exist_ok=True)
        self._load()

    def _load(self):
        if self._file.exists():
            data = json.loads(self._file.read_text())
            messages = [ChatMessage(**m) for m in data]
            self.chat_store.store_messages(str(self._file), messages)

    def _persist(self):
        messages = self.chat_store.get_messages(str(self._file))
        self._file.write_text(json.dumps([m.model_dump() for m in messages]))

    def put(self, message: ChatMessage) -> None:
        super().put(message)
        self._persist()

    def get(self, input: str = "") -> List[ChatMessage]:
        return super().get(input)

    def reset(self) -> None:
        super().reset()
        if self._file.exists():
            self._file.unlink()

Pick one backend and move on — the interface is identical.

Step 3: Wire memory into the chat engine

LlamaIndex’s CondensePlusContextChatEngine (or ContextChatEngine for simpler RAG) accepts a memory parameter. The memory instance you created in Step 2 plugs in directly.

from llama_index.core.chat_engine import CondensePlusContextChatEngine
from llama_index.core import VectorStoreIndex
from llama_index.llms.openai import OpenAI

# Assume you already have an index built from your data
index = VectorStoreIndex.from_documents(documents)

llm = OpenAI(model="gpt-4o-mini", temperature=0)

chat_engine = CondensePlusContextChatEngine.from_defaults(
    index=index,
    llm=llm,
    memory=memory,  # your Redis/Postgres/File memory instance
    system_prompt=(
        "You are a helpful assistant with access to the user's conversation history. "
        "Reference prior turns when relevant."
    ),
    verbose=True,
)

The engine now automatically:

1. Pulls recent history from memory.get() before each turn
2. Appends the new user message and assistant response via memory.put()
3. Your backend persists on every put()

Step 4: Persist and restore across sessions

The session identifier is your correlation key. In a web app, this typically maps to a user ID or conversation ID from your auth/session layer.

# FastAPI example
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    session_id: str
    user_id: str  # for authorization

class ChatResponse(BaseModel):
    response: str
    session_id: str

def get_memory(session_id: str, user_id: str) -> RedisChatMemory:
    # Validate user owns this session (pseudo-code)
    # if not session_belongs_to_user(session_id, user_id):
    #     raise HTTPException(403, "Session not found")
    return RedisChatMemory(
        redis_url="redis://localhost:6379",
        session_id=session_id,
        token_limit=3000,
    )

@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest, memory: RedisChatMemory = Depends(get_memory)):
    chat_engine = CondensePlusContextChatEngine.from_defaults(
        index=index,
        llm=llm,
        memory=memory,
        system_prompt="...",
    )
    response = chat_engine.chat(req.message)
    return ChatResponse(response=str(response), session_id=req.session_id)

New session flow:

1. Client sends session_id (generate UUID on first message if absent)
2. Server instantiates memory with that session_id
3. Memory loads existing history from backend (or starts empty)
4. Chat proceeds; every turn persists automatically

Resume session flow: Identical — the same session_id pulls prior context.

Step 5: Handle multi-user isolation

Never share a single memory instance across users. The session_id must be scoped to a user or conversation. Two patterns work:

Pattern A: Per-conversation (recommended)

# session_id = conversation_uuid
# Each conversation gets isolated history
memory = RedisChatMemory(redis_url, session_id=conversation_uuid)

Pattern B: Per-user with conversation segmentation

# session_id = f"user:{user_id}:conv:{conversation_id}"
# Allows listing a user's conversations via Redis SCAN
memory = RedisChatMemory(redis_url, session_id=f"user:{user_id}:conv:{conv_id}")

Add a metadata store (Postgres, DynamoDB) to track conversation titles, timestamps, and ownership — memory backends only store messages.

Step 6: Verify it works

Run this end-to-end test to confirm persistence survives process restarts.

# test_persistence.py
import uuid
from llama_index.core.llms import ChatMessage
from llama_index.llms.openai import OpenAI

# 1. Create a session
session_id = str(uuid.uuid4())
memory = RedisChatMemory("redis://localhost:6379", session_id=session_id, token_limit=3000)

llm = OpenAI(model="gpt-4o-mini")
chat_engine = CondensePlusContextChatEngine.from_defaults(
    index=index,
    llm=llm,
    memory=memory,
    system_prompt="You are a test assistant.",
)

# 2. First conversation turn
resp1 = chat_engine.chat("My favorite color is teal.")
print(f"Turn 1: {resp1}")

# 3. Second turn — should reference history
resp2 = chat_engine.chat("What did I just tell you?")
print(f"Turn 2: {resp2}")

# 4. Simulate process restart: new memory instance, same session_id
memory2 = RedisChatMemory("redis://localhost:6379", session_id=session_id, token_limit=3000)
chat_engine2 = CondensePlusContextChatEngine.from_defaults(
    index=index,
    llm=llm,
    memory=memory2,
    system_prompt="You are a test assistant.",
)

# 5. Third turn — must remember teal
resp3 = chat_engine2.chat("What's my favorite color?")
print(f"Turn 3 (after restart): {resp3}")

# 6. Assertions
assert "teal" in str(resp3).lower(), "Persistence failed: history not restored"
print("✅ Persistence verified")

Run it twice — once to populate, once after killing the process. The third response must contain “teal”.

Manual verification checklist

Common pitfalls

Token counting mismatch: If you pass a custom tokenizer_fn to ChatMemoryBuffer but your LLM uses a different tokenizer, the token_limit won’t match reality. Use the same tokenizer the model uses — for OpenAI models, tiktoken.encoding_for_model("gpt-4o-mini").encode.

Race conditions in Redis: The naive get → modify → set pattern loses updates under concurrent requests. For production, use Redis transactions (WATCH/MULTI/EXEC) or a Lua script that atomically appends. The RedisChatMemory above is single-threaded per session; if you scale horizontally, add a distributed lock or use Redis Streams.

Memory bloat: Unbounded conversations grow until they hit token_limit, then summarization kicks in (if enabled) or oldest messages drop. Monitor len(memory.get()) in production and alert if sessions routinely hit the cap — it means users are losing context.

Serialization drift: ChatMessage.model_dump() output changes across LlamaIndex versions. Pin your dependency version and test upgrades in staging. Consider storing a schema_version field in your backend for future migrations.

System prompt not persisted: The system prompt lives in the chat engine, not memory. If you change it between deployments, old sessions see the new prompt with old history — usually fine, but be aware.

Production hardening checklist

• Connection pooling: Reuse Redis/Postgres connections across requests (use redis.ConnectionPool or SQLAlchemy pool)
• Observability: Emit metrics — memory_load_latency_ms, memory_persist_latency_ms, session_message_count
• Graceful degradation: If Redis is down, fall back to in-memory ChatMemoryBuffer and log a warning (don’t fail the request)
• PII handling: If conversations contain sensitive data, encrypt at rest (Redis ACLs, Postgres pgcrypto, or application-layer encryption)
• Backup/restore: For file backend, include the memory directory in your backup strategy. For Redis/Postgres, standard DB backups cover it.

You now have a complete, production-ready pattern for persisting LlamaIndex chat history. The same memory instance works across any chat engine type — ContextChatEngine, CondenseQuestionChatEngine, or custom agents. Swap the backend without changing engine code.