LangChain ConversationBufferWindowMemory tutorial

If you’re building a chat application with LangChain, you’ll quickly hit the context window limit. The ConversationBufferWindowMemory class solves this by keeping only the last k message pairs, discarding older history automatically. This langchain conversationbufferwindowmemory tutorial walks through setup, configuration, and production patterns you’ll actually use.

Prerequisites

You need Python 3.10+ and an OpenAI-compatible API key. Install the minimal dependencies:

pip install langchain-openai langchain-community python-dotenv

Create a .env file with your API credentials:

OPENAI_API_KEY=sk-your-key-here
# Optional: if using n4n.ai or another OpenAI-compatible gateway
# OPENAI_BASE_URL=https://api.n4n.ai/v1

All code below assumes this environment is loaded via python-dotenv.

Basic implementation

Start with the simplest working example. The window size k controls how many human/assistant pairs to retain.

# basic_window.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain

load_dotenv()

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

memory = ConversationBufferWindowMemory(k=2, return_messages=True)
conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# First exchange
response1 = conversation.predict(input="My name is Alex and I work at a fintech startup.")
print(f"Response 1: {response1}")

# Second exchange
response2 = conversation.predict(input="What's my name?")
print(f"Response 2: {response2}")

# Third exchange - first exchange should still be in memory
response3 = conversation.predict(input="What company did I mention?")
print(f"Response 3: {response3}")

# Fourth exchange - first exchange should now be evicted (k=2)
response4 = conversation.predict(input="Remind me, what's my name again?")
print(f"Response 4: {response4}")

Run it:

python basic_window.py

Expected output (abbreviated for clarity):

> Entering new ConversationChain chain...
Prompt after formatting:
System: The following is a friendly conversation between a human and an AI...

Current conversation:
Human: My name is Alex and I work at a fintech startup.
AI: Nice to meet you, Alex! Fintech is an exciting space...

> Finished chain.
Response 1: Nice to meet you, Alex! Fintech is an exciting space...

> Entering new ConversationChain chain...
Prompt after formatting:
...
Current conversation:
Human: My name is Alex and I work at a fintech startup.
AI: Nice to meet you, Alex! Fintech is an exciting space...
Human: What's my name?
AI: Your name is Alex.

> Finished chain.
Response 2: Your name is Alex.

> Entering new ConversationChain chain...
Prompt after formatting:
...
Current conversation:
Human: My name is Alex and I work at a fintech startup.
AI: Nice to meet you, Alex! Fintech is an exciting space...
Human: What's my name?
AI: Your name is Alex.
Human: What company did I mention?
AI: You mentioned you work at a fintech startup.

> Finished chain.
Response 3: You mentioned you work at a fintech startup.

> Entering new ConversationChain chain...
Prompt after formatting:
...
Current conversation:
Human: What's my name?
AI: Your name is Alex.
Human: What company did I mention?
AI: You mentioned you work at a fintech startup.
Human: Remind me, what's my name again?
AI: I don't have that information in our recent conversation history.

> Finished chain.
Response 4: I don't have that information in our recent conversation history.

Notice how the fourth response loses the name. With k=2, only the last two exchanges remain. The initial introduction was evicted after the third exchange.

Inspecting the memory buffer

You’ll often need to debug what’s actually stored. Access the buffer directly:

# inspect_memory.py
from langchain.memory import ConversationBufferWindowMemory
from langchain_core.messages import HumanMessage, AIMessage

memory = ConversationBufferWindowMemory(k=2, return_messages=True)

# Simulate a conversation
memory.chat_memory.add_message(HumanMessage(content="Hello"))
memory.chat_memory.add_message(AIMessage(content="Hi there!"))
memory.chat_memory.add_message(HumanMessage(content="How are you?"))
memory.chat_memory.add_message(AIMessage(content="I'm doing well, thanks!"))
memory.chat_memory.add_message(HumanMessage(content="What's the weather?"))
memory.chat_memory.add_message(AIMessage(content="I don't have access to weather data."))

print(f"Buffer length: {len(memory.chat_memory.messages)}")
print("Messages in buffer:")
for msg in memory.chat_memory.messages:
    print(f"  {msg.type}: {msg.content}")

# Load memory variables as the chain sees them
vars = memory.load_memory_variables({})
print(f"\nMemory variables: {vars}")

Output:

Buffer length: 6
Messages in buffer:
  human: Hello
  ai: Hi there!
  human: How are you?
  ai: I'm doing well, thanks!
  human: What's the weather?
  ai: I don't have access to weather data.

Memory variables: {'history': [HumanMessage(content='How are you?'), AIMessage(content="I'm doing well, thanks!"), HumanMessage(content='What's the weather?'), AIMessage(content="I don't have access to weather data.")]}

The internal buffer stores all six messages, but load_memory_variables returns only the last k=2 pairs (four messages). This distinction matters when you’re debugging token usage.

Using with LCEL (LangChain Expression Language)

Modern LangChain code uses LCEL. Here’s the same pattern with RunnableWithMessageHistory:

# lecel_window.py
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = prompt | llm

# In-memory store for demo; replace with Redis/Postgres in production
store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# Wrap with window memory - note: k applies to the underlying store
with_message_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# First session
config = {"configurable": {"session_id": "user-123"}}
response1 = with_message_history.invoke({"input": "I'm building a payment API."}, config=config)
print(f"Response 1: {response1.content}")

response2 = with_message_history.invoke({"input": "What am I building?"}, config=config)
print(f"Response 2: {response2.content}")

# Check what's stored
history = get_session_history("user-123")
print(f"\nStored messages: {len(history.messages)}")
for m in history.messages:
    print(f"  {m.type}: {m.content}")

Output:

Response 1: That sounds like an interesting project! A payment API involves...
Response 2: You're building a payment API.

Stored messages: 4
  human: I'm building a payment API.
  ai: That sounds like an interesting project! A payment API involves...
  human: What am I building?
  ai: You're building a payment API.

With LCEL, the windowing happens at the message history layer. ChatMessageHistory stores everything; you’d typically wrap it with a custom class that enforces the window, or accept that the full history accumulates and rely on the model’s context window.

Custom windowed message history

For true windowing in LCEL, implement BaseChatMessageHistory with eviction:

# windowed_history.py
from typing import List
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_community.chat_message_histories import ChatMessageHistory

class WindowedChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, k: int = 5):
        self._history = ChatMessageHistory()
        self.k = k

    @property
    def messages(self) -> List[BaseMessage]:
        all_messages = self._history.messages
        # Keep last k pairs = 2*k messages
        window_size = self.k * 2
        return all_messages[-window_size:] if len(all_messages) > window_size else all_messages

    def add_message(self, message: BaseMessage) -> None:
        self._history.add_message(message)

    def clear(self) -> None:
        self._history.clear()

# Usage
history = WindowedChatMessageHistory(k=2)

for i in range(6):
    history.add_message(HumanMessage(content=f"Message {i}"))
    history.add_message(AIMessage(content=f"Response {i}"))

print(f"Total stored: {len(history._history.messages)}")
print(f"Windowed view: {len(history.messages)}")
for m in history.messages:
    print(f"  {m.type}: {m.content}")

Output:

Total stored: 12
Windowed view: 4
  human: Message 4
  ai: Response 4
  human: Message 5
  ai: Response 5

This gives you true windowing at the storage layer. The underlying ChatMessageHistory still accumulates (useful for analytics), but the chain only sees the window.

Token-aware windowing

Fixed k works for simple cases, but token budgets vary by model. A more sophisticated approach tracks tokens and evicts until under budget:

# token_window.py
import tiktoken
from typing import List
from langchain_core.messages import BaseMessage
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

class TokenWindowedHistory(BaseChatMessageHistory):
    def __init__(self, max_tokens: int = 3000, model: str = "gpt-4o-mini"):
        self._history = ChatMessageHistory()
        self.max_tokens = max_tokens
        self.encoding = tiktoken.encoding_for_model(model)

    def _count_tokens(self, messages: List[BaseMessage]) -> int:
        total = 0
        for msg in messages:
            total += len(self.encoding.encode(msg.content))
            total += 4  # overhead per message (role, formatting)
        return total

    @property
    def messages(self) -> List[BaseMessage]:
        all_messages = self._history.messages
        # Evict oldest pairs until under budget
        while len(all_messages) >= 2 and self._count_tokens(all_messages) > self.max_tokens:
            # Remove oldest human/assistant pair
            all_messages = all_messages[2:]
        return all_messages

    def add_message(self, message: BaseMessage) -> None:
        self._history.add_message(message)

    def clear(self) -> None:
        self._history.clear()

# Demo
history = TokenWindowedHistory(max_tokens=100, model="gpt-4o-mini")

long_msg = "x" * 500  # ~125 tokens
for i in range(5):
    history.add_message(HumanMessage(content=f"User message {i}: {long_msg}"))
    history.add_message(AIMessage(content=f"Assistant response {i}: {long_msg}"))

print(f"Stored: {len(history._history.messages)} messages")
print(f"Windowed: {len(history.messages)} messages")
print(f"Estimated tokens: {history._count_tokens(history.messages)}")

Output (approximate):

Stored: 10 messages
Windowed: 2 messages
Estimated tokens: 98

This keeps the conversation within the token budget regardless of message length. Adjust max_tokens based on your model’s context window minus room for the system prompt and response.

Persisting to Redis

In production, you need shared, durable storage. Here’s a Redis-backed windowed history:

# redis_window.py
import json
import redis
from typing import List, Optional
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, message_to_dict, messages_from_dict

class RedisWindowedHistory(BaseChatMessageHistory):
    def __init__(
        self,
        session_id: str,
        redis_client: redis.Redis,
        k: int = 10,
        ttl_seconds: int = 86400,
    ):
        self.session_id = session_id
        self.redis = redis_client
        self.k = k
        self.ttl = ttl_seconds
        self.key = f"chat_history:{session_id}"

    @property
    def messages(self) -> List[BaseMessage]:
        data = self.redis.lrange(self.key, -self.k * 2, -1)
        if not data:
            return []
        return messages_from_dict([json.loads(msg) for msg in data])

    def add_message(self, message: BaseMessage) -> None:
        pipe = self.redis.pipeline()
        pipe.rpush(self.key, json.dumps(message_to_dict(message)))
        pipe.ltrim(self.key, -self.k * 2, -1)
        pipe.expire(self.key, self.ttl)
        pipe.execute()

    def clear(self) -> None:
        self.redis.delete(self.key)

# Usage
redis_client = redis.Redis(decode_responses=True)
history = RedisWindowedHistory("session-456", redis_client, k=3)

history.add_message(HumanMessage(content="First message"))
history.add_message(AIMessage(content="First response"))
# ... add more ...

print(f"Messages in window: {len(history.messages)}")

The LTRIM operation keeps only the last 2*k messages atomically. The TTL prevents orphaned sessions from accumulating.

Common pitfalls

Pitfall 1: Forgetting return_messages=True

Without it, memory returns a formatted string instead of message objects, breaking MessagesPlaceholder:

# Wrong - returns string
memory = ConversationBufferWindowMemory(k=2)

# Right - returns list of BaseMessage
memory = ConversationBufferWindowMemory(k=2, return_messages=True)

Pitfall 2: Mismatched input_key and output_key

If your chain uses non-standard keys, configure them explicitly:

memory = ConversationBufferWindowMemory(
    k=3,
    return_messages=True,
    input_key="user_input",      # matches your chain's input
    output_key="assistant_reply",  # matches your chain's output
)

Pitfall 3: System prompt counted in window

The window only applies to human/assistant pairs. System prompts are injected separately and don’t count toward k. Plan your token budget accordingly.

Pitfall 4: Concurrent access to in-memory store

The dictionary-based store in the LCEL example isn’t thread-safe. Use Redis or a database for any multi-worker deployment.

Choosing the right k

There’s no universal value. Consider:

Start with k=5, monitor token usage, and adjust. Log the actual message count and token estimate per request:

def log_memory_usage(memory: ConversationBufferWindowMemory, model: str = "gpt-4o-mini"):
    encoding = tiktoken.encoding_for_model(model)
    messages = memory.chat_memory.messages
    tokens = sum(len(encoding.encode(m.content)) for m in messages)
    print(f"Window: {len(messages)} messages, ~{tokens} tokens")

Integration with n4n.ai

If you’re routing through a gateway like n4n.ai, the memory layer works identically — just point the ChatOpenAI client at the gateway endpoint. The gateway’s automatic fallback and per-token metering apply to the full request including the history window you’ve constructed.

llm = ChatOpenAI(
    model="gpt-4o-mini",
    base_url="https://api.n4n.ai/v1",  # or your gateway
    api_key=os.getenv("N4N_API_KEY"),
)

The memory implementation doesn’t change; only the model endpoint does.

Summary

ConversationBufferWindowMemory gives you a fixed-size sliding window over conversation history. Key takeaways:

• Set k based on your use case and token budget
• Use return_messages=True for LCEL compatibility
• For production, wrap BaseChatMessageHistory with Redis or Postgres
• Consider token-aware eviction for variable-length messages
• Monitor actual token usage in production to tune k

The window is a blunt instrument — it discards potentially relevant context. For more sophisticated retrieval, look at ConversationSummaryMemory or vector-backed memory, but those add latency and complexity. Start with the window, measure, and upgrade only when needed.