Token limits and chat memory truncation in LlamaIndex

Token limits are the constraint every LLM application hits first. Whether you’re building a customer support bot or a coding assistant, the conversation eventually exceeds the model’s context window. This llamaindex token limit memory truncation tutorial walks through the strategies LlamaIndex provides, where they break down, and how to build a truncation pipeline that survives production traffic.

Understanding the constraint

Every model has a hard context window — 4K for older models, 128K for GPT-4o, 200K for Claude 3.5 Sonnet, 1M for Gemini 1.5 Pro. That window must hold the system prompt, retrieved context, conversation history, and the current user message plus completion. When the total exceeds the limit, the API returns an error.

LlamaIndex doesn’t automatically solve this. The framework provides memory classes and token counters, but the truncation policy is yours to define. Most teams start with ChatMemoryBuffer and discover its limits under load.

from llama_index.core.memory import ChatMemoryBuffer
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o-mini")
memory = ChatMemoryBuffer.from_defaults(token_limit=3000, llm=llm)

This allocates 3,000 tokens for history. The rest of the context window goes to system prompt, RAG context, and completion. But ChatMemoryBuffer uses a simple FIFO eviction — oldest messages drop first. That loses critical context like user preferences or earlier decisions.

Memory types and their tradeoffs

LlamaIndex ships with three memory implementations. Each makes different assumptions about what matters.

ChatMemoryBuffer — token-aware FIFO

from llama_index.core.memory import ChatMemoryBuffer
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o")
memory = ChatMemoryBuffer.from_defaults(
    token_limit=8000,
    llm=llm,
    chat_store_key="user_123"
)

Pros: Accurate token counting via the LLM’s tokenizer. Simple API. Persists to ChatStore backends (Redis, Postgres, in-memory).

Cons: Blind eviction. No concept of message importance. System prompt and tool calls count against the limit unless you exclude them manually.

ChatSummaryMemoryBuffer — compress instead of drop

from llama_index.core.memory import ChatSummaryMemoryBuffer
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o")
memory = ChatSummaryMemoryBuffer.from_defaults(
    token_limit=8000,
    llm=llm,
    tokenizer_fn=llm.tokenizer.encode
)

When the buffer exceeds token_limit, it summarizes the oldest messages into a single summary message using the LLM. The summary stays in history.

Pros: Preserves semantic content of long conversations. Better for multi-session contexts.

Cons: Adds latency (extra LLM call on each truncation). Summary quality varies. Token counting on the summary itself is approximate. Can create summary-of-summary drift over very long conversations.

VectorStoreMemory — semantic retrieval

from llama_index.core.memory import VectorStoreMemory
from llama_index.vector_stores.redis import RedisVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding

vector_store = RedisVectorStore(index_name="chat_memory")
embed_model = OpenAIEmbedding(model="text-embedding-3-small")

memory = VectorStoreMemory.from_defaults(
    vector_store=vector_store,
    embed_model=embed_model,
    retriever_kwargs={"similarity_top_k": 5},
    token_limit=4000
)

Stores every message as a vector. On each turn, retrieves the most relevant past messages within the token budget.

Pros: Recovers relevant context regardless of recency. Scales to arbitrarily long histories.

Cons: Retrieval adds latency. Relevance ≠ necessity — may miss critical but semantically dissimilar context (e.g., “my API key is sk-…”). Requires a vector store dependency. Token limit applies to retrieved set, not total history.

Building a production truncation pipeline

Real applications need deterministic behavior. The following pattern combines explicit priority tiers with token budgeting — no surprise LLM calls, no silent data loss.

Step 1: Define message priority

from enum import IntEnum
from dataclasses import dataclass
from typing import List, Optional
from llama_index.core.llms import ChatMessage, MessageRole

class Priority(IntEnum):
    SYSTEM = 0      # Never drop
    TOOL_RESULT = 1 # Expensive to recompute
    USER = 2        # Core conversation
    ASSISTANT = 3   # Can be summarized
    SUMMARY = 4     # Already compressed

@dataclass
class PrioritizedMessage:
    message: ChatMessage
    priority: Priority
    tokens: int
    metadata: dict = None

Step 2: Accurate token counting

import tiktoken
from llama_index.llms.openai import OpenAI

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    """Count tokens using the model's actual tokenizer."""
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        encoding = tiktoken.get_encoding("cl100k_base")
    return len(encoding.encode(text))

def count_message_tokens(message: ChatMessage, model: str = "gpt-4o") -> int:
    """Count tokens for a ChatMessage including role overhead."""
    # Role tokens: ~3-4 per message depending on model
    role_overhead = 4
    content_tokens = count_tokens(message.content or "", model)
    return role_overhead + content_tokens

Step 3: Budget allocator

from typing import List

class TokenBudget:
    def __init__(
        self,
        context_window: int,
        system_prompt_tokens: int,
        rag_budget: int,
        completion_budget: int,
        safety_margin: int = 100
    ):
        self.context_window = context_window
        self.available = (
            context_window
            - system_prompt_tokens
            - rag_budget
            - completion_budget
            - safety_margin
        )
    
    def allocate(self, messages: List[PrioritizedMessage]) -> List[PrioritizedMessage]:
        """Return messages that fit, highest priority first."""
        # Sort by priority (lower = more important), then by recency (newer first)
        sorted_msgs = sorted(
            messages,
            key=lambda m: (m.priority, -m.metadata.get("turn_index", 0))
        )
        
        selected = []
        used = 0
        
        for msg in sorted_msgs:
            if used + msg.tokens <= self.available:
                selected.append(msg)
                used += msg.tokens
            else:
                # Log what we're dropping for observability
                print(f"Dropping {msg.priority.name} message: {msg.message.content[:50]}...")
        
        # Restore chronological order
        selected.sort(key=lambda m: m.metadata.get("turn_index", 0))
        return selected

Step 4: Integrate with LlamaIndex chat engine

from llama_index.core.chat_engine import CondensePlusContextChatEngine
from llama_index.core.memory import BaseMemory
from llama_index.core.llms import ChatMessage
from typing import List, Any

class PrioritizedMemory(BaseMemory):
    """Memory that enforces priority-based truncation."""
    
    def __init__(
        self,
        token_budget: TokenBudget,
        llm: OpenAI,
        chat_store_key: str = "default"
    ):
        self.token_budget = token_budget
        self.llm = llm
        self.chat_store_key = chat_store_key
        self._messages: List[PrioritizedMessage] = []
        self._turn_index = 0
    
    def get(self, input: str = "", **kwargs) -> List[ChatMessage]:
        allocated = self.token_budget.allocate(self._messages)
        return [m.message for m in allocated]
    
    def put(self, message: ChatMessage) -> None:
        priority = self._infer_priority(message)
        tokens = count_message_tokens(message, self.llm.model)
        
        self._messages.append(PrioritizedMessage(
            message=message,
            priority=priority,
            tokens=tokens,
            metadata={"turn_index": self._turn_index}
        ))
        self._turn_index += 1
    
    def _infer_priority(self, message: ChatMessage) -> Priority:
        if message.role == MessageRole.SYSTEM:
            return Priority.SYSTEM
        elif message.role == MessageRole.TOOL:
            return Priority.TOOL_RESULT
        elif message.role == MessageRole.USER:
            return Priority.USER
        elif message.additional_kwargs.get("tool_calls"):
            return Priority.TOOL_RESULT
        else:
            return Priority.ASSISTANT
    
    def reset(self) -> None:
        self._messages.clear()
        self._turn_index = 0

Step 5: Wire it up

from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.chat_engine import CondensePlusContextChatEngine
from llama_index.core.retrievers import VectorIndexRetriever

# Configure
llm = OpenAI(model="gpt-4o", temperature=0)
embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.llm = llm
Settings.embed_model = embed_model

# Budget: 128K window - 2K system - 20K RAG - 4K completion - 100 margin = ~101.9K for history
budget = TokenBudget(
    context_window=128000,
    system_prompt_tokens=2000,
    rag_budget=20000,
    completion_budget=4000
)

memory = PrioritizedMemory(token_budget=budget, llm=llm)

# Build chat engine with your retriever
retriever = VectorIndexRetriever(index=your_index, similarity_top_k=5)

chat_engine = CondensePlusContextChatEngine.from_defaults(
    retriever=retriever,
    memory=memory,
    llm=llm,
    context_prompt=(
        "Relevant context:\n{context_str}\n\n"
        "Answer the user's question using this context."
    ),
    verbose=True
)

# Use it
response = chat_engine.chat("What's my API key?")
print(response.response)

Common pitfalls

Pitfall 1: Counting tokens incorrectly

ChatMemoryBuffer uses the LLM’s tokenizer property. But OpenAI’s Python SDK tokenizer doesn’t match the API exactly for all models. Use tiktoken directly with the correct encoding:

# Wrong — uses default cl100k_base for everything
len(llm.tokenizer.encode(text))

# Right — model-specific encoding
encoding = tiktoken.encoding_for_model("gpt-4o")
len(encoding.encode(text))

For non-OpenAI models, you need the provider’s tokenizer. Anthropic, Google, and open models each have different tokenization. If you route across providers, count tokens per-provider or use a conservative estimator.

Pitfall 2: Ignoring tool call overhead

Tool calls and their results consume significant tokens. A single function call with a large JSON response can exceed 2K tokens. The priority system above treats TOOL_RESULT as high priority, but you may need to truncate the content of tool results (e.g., truncate a 50KB API response to 2KB summary) before it enters memory.

def truncate_tool_result(content: str, max_tokens: int = 1000) -> str:
    tokens = count_tokens(content)
    if tokens <= max_tokens:
        return content
    # Keep first and last portions, note truncation
    encoding = tiktoken.encoding_for_model("gpt-4o")
    ids = encoding.encode(content)
    keep = max_tokens // 2
    truncated = encoding.decode(ids[:keep]) + "\n... [truncated] ...\n" + encoding.decode(ids[-keep:])
    return truncated

Pitfall 3: System prompt bloat

Teams add instructions, few-shot examples, and context to the system prompt until it consumes 20% of the window. Move static context to RAG. Keep system prompts under 1,500 tokens. If you need more, you’re designing a different architecture.

Pitfall 4: No observability on truncation

You won’t know truncation is hurting quality until users complain. Log every truncation event:

import structlog

logger = structlog.get_logger()

def log_truncation(dropped_messages: List[PrioritizedMessage], budget: TokenBudget):
    logger.warning(
        "memory_truncation",
        dropped_count=len(dropped_messages),
        dropped_tokens=sum(m.tokens for m in dropped_messages),
        available_tokens=budget.available,
        priorities=[m.priority.name for m in dropped_messages]
    )

Pitfall 5: Assuming ChatMemoryBuffer handles multi-turn tool loops

It doesn’t. A ReAct agent making 5 tool calls in one turn generates 10+ messages (call + result pairs). The buffer counts all of them. Either increase the token limit or summarize tool loops after execution.

Advanced: Summarization as a background job

For very long conversations, inline summarization adds unacceptable latency. Move it to a background worker:

# In your chat handler — synchronous, fast
def handle_message(user_input: str):
    # 1. Get current memory (already truncated to budget)
    history = memory.get()
    
    # 2. Run chat engine
    response = chat_engine.chat(user_input)
    
    # 3. Store new messages
    memory.put(ChatMessage(role=MessageRole.USER, content=user_input))
    memory.put(ChatMessage(role=MessageRole.ASSISTANT, content=response.response))
    
    # 4. Check if summarization needed (async, fire-and-forget)
    if memory.needs_summarization():
        summarize_task.delay(memory.chat_store_key)
    
    return response

# Background worker — can use a cheaper model
@celery.task
def summarize_task(chat_store_key: str):
    messages = chat_store.get_messages(chat_store_key)
    if len(messages) < 20:  # Threshold
        return
    
    # Summarize oldest 10 messages
    to_summarize = messages[:10]
    summary_prompt = f"Summarize this conversation preserving key facts:\n{format_messages(to_summarize)}"
    summary = cheap_llm.complete(summary_prompt)
    
    # Replace with summary message
    chat_store.delete_messages(chat_store_key, to_summarize)
    chat_store.add_message(chat_store_key, ChatMessage(
        role=MessageRole.SYSTEM,
        content=f"Conversation summary: {summary.text}",
        additional_kwargs={"priority": Priority.SUMMARY}
    ))

This keeps the hot path fast while bounding memory growth. The summary message carries Priority.SUMMARY so it’s the first to go if pressure continues.

Routing across models with different windows

If your application routes requests to different models (e.g., simple queries to GPT-4o-mini, complex to Claude 3.5 Sonnet), the context window changes per request. Your memory layer must either:

1. Maintain per-model budgets — store the full history, compute a different truncation per-request
2. Normalize to the smallest window — safe but wastes capacity on larger models
3. Use a gateway that exposes the active model’s limits — n4n.ai forwards provider cache-control hints and model metadata so your truncation logic can adapt dynamically

# Example: dynamic budget from gateway response headers
def get_budget_for_model(model: str, gateway_headers: dict) -> TokenBudget:
    # Gateway returns x-model-context-window: 128000
    context_window = int(gateway_headers.get("x-model-context-window", 128000))
    return TokenBudget(
        context_window=context_window,
        system_prompt_tokens=2000,
        rag_budget=20000,
        completion_budget=4000
    )

Testing your truncation logic

Write property-based tests that verify invariants:

import pytest
from hypothesis import given, strategies as st

@given(st.lists(st.text(min_size=1, max_size=500), min_size=1, max_size=50))
def test_budget_never_exceeded(messages_content):
    budget = TokenBudget(128000, 2000, 20000, 4000)
    messages = [
        PrioritizedMessage(
            message=ChatMessage(role=MessageRole.USER, content=c),
            priority=Priority.USER,
            tokens=count_tokens(c),
            metadata={"turn_index": i}
        )
        for i, c in enumerate(messages_content)
    ]
    
    allocated = budget.allocate(messages)
    total_tokens = sum(m.tokens for m in allocated)
    
    assert total_tokens <= budget.available

def test_system_messages_never_dropped():
    budget = TokenBudget(4000, 500, 1000, 1000)  # Very tight
    messages = [
        PrioritizedMessage(
            message=ChatMessage(role=MessageRole.SYSTEM, content="x" * 3000),
            priority=Priority.SYSTEM,
            tokens=3000,
            metadata={"turn_index": 0}
        ),
        PrioritizedMessage(
            message=ChatMessage(role=MessageRole.USER, content="hello"),
            priority=Priority.USER,
            tokens=10,
            metadata={"turn_index": 1}
        )
    ]
    
    allocated = budget.allocate(messages)
    # System message fits alone, user message dropped
    assert len(allocated) == 1
    assert allocated[0].priority == Priority.SYSTEM

Summary

Token limits are a hard constraint, not a suggestion. LlamaIndex gives you building blocks — ChatMemoryBuffer, ChatSummaryMemoryBuffer, VectorStoreMemory — but no single class handles production requirements. Build a priority-based budget allocator, count tokens with tiktoken per model, log every truncation, and move summarization off the hot path. Test the allocator with property-based tests. If you route across models, make the budget dynamic.

The patterns above have kept chat applications stable at millions of messages per day. Adapt the priority tiers to your domain — medical apps might prioritize tool results higher, coding assistants might prioritize recent code context. The framework is the same: explicit priorities, accurate counting, observable truncation.