LangChain memory for multi-user chat applications

LangChain memory works fine for single-user demos. In multi-user production systems, the same patterns leak context across sessions, balloon latency, and crash under concurrent load. This guide walks through the decisions that separate a prototype from a system that survives real traffic.

The core problem: memory is global by default

LangChain’s ConversationBufferMemory and its siblings store history in memory attached to a chain instance. If you instantiate one chain per request but share a single memory object — or worse, use a global singleton — every user sees every other user’s conversation. The fix is not “be careful.” The fix is architectural: memory must be scoped to a session identifier and backed by a store that survives process restarts.

# WRONG: global memory shared across all users
memory = ConversationBufferMemory(return_messages=True)
chain = ConversationChain(llm=llm, memory=memory)

# Request handler
async def chat(request):
    return chain.predict(input=request.message)  # all users share history

# RIGHT: memory scoped to session, backed by external store
async def chat(request):
    session_id = request.session_id  # from auth token, cookie, or header
    memory = await get_memory_for_session(session_id)
    chain = ConversationChain(llm=llm, memory=memory)
    return chain.predict(input=request.message)

Choose the right memory type for the job

LangChain ships several memory classes. For multi-user chat, only three matter in practice.

ConversationBufferMemory

Stores every message. Simple, predictable, grows without bound. Use when conversations are short (under 10-15 turns) or when you need full fidelity for compliance.

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    return_messages=True,
    memory_key="history",
    input_key="input",
    output_key="output",
)

ConversationBufferWindowMemory

Keeps only the last k exchanges. Bounded memory footprint, drops old context. The pragmatic default for most chat applications.

from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(
    k=10,  # last 10 human/ai pairs
    return_messages=True,
)

ConversationSummaryBufferMemory

Summarizes older turns while keeping recent messages verbatim. Best for long-running conversations where you need both detail and breadth. Adds an LLM call on every write — factor that into latency and cost.

from langchain.memory import ConversationSummaryBufferMemory

memory = ConversationSummaryBufferMemory(
    llm=llm,  # separate, cheaper model for summarization
    max_token_limit=2000,
    return_messages=True,
)

Tradeoff: Summary memory introduces a failure mode — if the summarization LLM hallucinates, you lose ground truth. Buffer window is safer for high-stakes domains (medical, legal, financial).

Session isolation: where the session id comes from

The session identifier must be immutable for the conversation’s lifetime and unguessable. Three patterns work:

1. Authenticated users — derive from user ID + conversation ID

def get_session_id(user_id: str, conversation_id: str) -> str:
    return f"user:{user_id}:conv:{conversation_id}"

2. Anonymous users — cryptographically random token

import secrets

def create_anonymous_session() -> str:
    return f"anon:{secrets.token_urlsafe(32)}"

3. Stateless — embed in client-signed JWT

import jwt

def create_session_token(user_id: str, conversation_id: str, secret: str) -> str:
    payload = {"uid": user_id, "cid": conversation_id, "exp": time.time() + 86400}
    return jwt.encode(payload, secret, algorithm="HS256")

def parse_session_token(token: str, secret: str) -> tuple[str, str]:
    payload = jwt.decode(token, secret, algorithms=["HS256"])
    return payload["uid"], payload["cid"]

Never accept a session ID directly from the client without validation. An attacker who guesses or steals a session ID reads that user’s history.

Storage backends: pick one and instrument it

In-memory dictionaries ({}) work for local development. They fail the moment you run multiple replicas, restart a pod, or exceed RAM. Production needs an external store.

Redis: the default choice

Sub-millisecond latency, built-in TTL, horizontal scaling via Cluster. Use redis-py with connection pooling.

import redis.asyncio as redis
from langchain.memory import ConversationBufferWindowMemory
from langchain.schema import messages_from_dict, messages_to_dict

class RedisMemory(ConversationBufferWindowMemory):
    def __init__(self, session_id: str, redis_client: redis.Redis, ttl: int = 86400, k: int = 10):
        super().__init__(k=k, return_messages=True)
        self.session_id = session_id
        self.redis = redis_client
        self.ttl = ttl
        self.key = f"chat:memory:{session_id}"

    async def load_memory_variables(self, inputs: dict) -> dict:
        data = await self.redis.get(self.key)
        if data:
            messages = messages_from_dict(json.loads(data))
            self.chat_memory.messages = messages
        return {"history": self.chat_memory.messages}

    async def save_context(self, inputs: dict, outputs: dict) -> None:
        await super().save_context(inputs, outputs)
        data = json.dumps(messages_to_dict(self.chat_memory.messages))
        await self.redis.setex(self.key, self.ttl, data)

Pitfall: Redis memory grows with every conversation. Set a TTL (24-72 hours typical) and monitor used_memory_human. Evict stale sessions proactively.

PostgreSQL: when you need queryability

If you need to search conversations, join with user tables, or run analytics, Postgres wins. Use asyncpg or SQLAlchemy async. Store messages as JSONB.

CREATE TABLE chat_memory (
    session_id TEXT PRIMARY KEY,
    user_id TEXT NOT NULL,
    messages JSONB NOT NULL DEFAULT '[]',
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    expires_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX idx_chat_memory_user ON chat_memory(user_id);
CREATE INDEX idx_chat_memory_expires ON chat_memory(expires_at);

import asyncpg
import json
from datetime import datetime, timedelta

class PostgresMemory(ConversationBufferWindowMemory):
    def __init__(self, session_id: str, pool: asyncpg.Pool, k: int = 10):
        super().__init__(k=k, return_messages=True)
        self.session_id = session_id
        self.pool = pool

    async def load_memory_variables(self, inputs: dict) -> dict:
        row = await self.pool.fetchrow(
            "SELECT messages FROM chat_memory WHERE session_id = $1", self.session_id
        )
        if row:
            self.chat_memory.messages = messages_from_dict(json.loads(row["messages"]))
        return {"history": self.chat_memory.messages}

    async def save_context(self, inputs: dict, outputs: dict) -> None:
        await super().save_context(inputs, outputs)
        messages_json = json.dumps(messages_to_dict(self.chat_memory.messages))
        expires = datetime.utcnow() + timedelta(days=7)
        await self.pool.execute(
            """
            INSERT INTO chat_memory (session_id, messages, expires_at)
            VALUES ($1, $2, $3)
            ON CONFLICT (session_id) DO UPDATE SET messages = $2, updated_at = NOW(), expires_at = $3
            """,
            self.session_id, messages_json, expires
        )

Pitfall: JSONB serialization adds ~1-2ms per request. Connection pooling is mandatory — configure min_size=10, max_size=50 for typical workloads.

DynamoDB / Cosmos DB / Firestore: serverless fits

If your stack is serverless, use the native document store. Same pattern: session ID as partition key, messages as list attribute, TTL attribute for auto-expiry. Latency is higher (5-15ms) but ops burden is near zero.

Concurrency: the lost-update problem

Two requests for the same session arrive simultaneously. Both load memory, both append, both write. One write wins; the other’s turn vanishes.

Solution 1: Optimistic locking with version field

# Add version column to Postgres table
# ALTER TABLE chat_memory ADD COLUMN version INT DEFAULT 0;

async def save_context(self, inputs: dict, outputs: dict) -> None:
    await super().save_context(inputs, outputs)
    messages_json = json.dumps(messages_to_dict(self.chat_memory.messages))
    for attempt in range(3):
        result = await self.pool.execute(
            """
            UPDATE chat_memory
            SET messages = $2, version = version + 1, updated_at = NOW()
            WHERE session_id = $1 AND version = $3
            """,
            self.session_id, messages_json, self._version
        )
        if result == "UPDATE 1":
            self._version += 1
            return
        # Reload and retry
        row = await self.pool.fetchrow("SELECT messages, version FROM chat_memory WHERE session_id = $1", self.session_id)
        self.chat_memory.messages = messages_from_dict(json.loads(row["messages"]))
        self._version = row["version"]
    raise ConcurrentModificationError("Failed to save after retries")

Solution 2: Redis Lua script for atomic append

APPEND_SCRIPT = """
local key = KEYS[1]
local msg = ARGV[1]
local max_len = tonumber(ARGV[2])
local ttl = tonumber(ARGV[3])
redis.call('RPUSH', key, msg)
redis.call('LTRIM', key, -max_len, -1)
redis.call('EXPIRE', key, ttl)
return redis.call('LRANGE', key, 0, -1)
"""

async def save_context(self, inputs: dict, outputs: dict) -> None:
    # Serialize the new messages only
    new_messages = messages_to_dict(self.chat_memory.messages[-2:])  # human + ai
    for msg in new_messages:
        await self.redis.eval(APPEND_SCRIPT, 1, self.key, json.dumps(msg), self.k * 2, self.ttl)

Solution 3: Single-writer per session via distributed lock

# Redis SET NX with short TTL
lock_acquired = await self.redis.set(f"lock:{self.session_id}", "1", nx=True, ex=5)
if not lock_acquired:
    await asyncio.sleep(0.1)
    return await self.save_context(inputs, outputs)  # retry
try:
    await self._do_save()
finally:
    await self.redis.delete(f"lock:{self.session_id}")

Optimistic locking (Solution 1) scales best. Distributed locks serialize throughput — avoid unless contention is genuinely rare.

Token budgeting: prevent context window overflow

Memory feeds the prompt. If history + system prompt + user input exceeds the model’s context window, the request fails or truncates silently. Budget tokens explicitly.

import tiktoken

class TokenBudgetMemory(ConversationBufferWindowMemory):
    def __init__(self, llm, max_tokens: int = 3000, **kwargs):
        super().__init__(**kwargs)
        self.llm = llm
        self.max_tokens = max_tokens
        self.encoding = tiktoken.encoding_for_model(llm.model_name)

    def _count_tokens(self, messages: list) -> int:
        return sum(len(self.encoding.encode(m.content)) for m in messages)

    def load_memory_variables(self, inputs: dict) -> dict:
        messages = self.chat_memory.messages
        # Trim from oldest until under budget
        while messages and self._count_tokens(messages) > self.max_tokens:
            messages.pop(0)
        return {"history": messages}

Rule of thumb: Reserve 30-40% of the context window for the system prompt and expected output. For a 4k model, keep history under ~2500 tokens. For 128k models, the budget is less binding but latency grows with context — trim anyway.

Streaming and memory: don’t double-write

When streaming tokens, the final assembled message must be written to memory exactly once. A common bug: writing partial chunks on each callback, then writing the full message again at the end.

# WRONG: writes partial chunks + final message
async def on_llm_new_token(token: str):
    await memory.save_context({"input": user_input}, {"output": token})  # called per token

# RIGHT: accumulate, write once
class StreamingMemoryCallback:
    def __init__(self, memory, user_input):
        self.memory = memory
        self.user_input = user_input
        self.buffer = []

    async def on_llm_new_token(self, token: str):
        self.buffer.append(token)

    async def on_llm_end(self, response):
        full_output = "".join(self.buffer)
        await self.memory.save_context({"input": self.user_input}, {"output": full_output})

Observability: you can’t debug what you don’t measure

Instrument every memory operation. At minimum, log:

• Session ID (hashed for privacy)
• Memory load latency (p50, p95, p99)
• Memory save latency
• Message count per session
• Token count per session
• Error rates by type (timeout, serialization, concurrency conflict)

import time
from prometheus_client import Histogram, Counter

MEMORY_LOAD_LATENCY = Histogram("memory_load_seconds", "Memory load latency", ["backend"])
MEMORY_SAVE_LATENCY = Histogram("memory_save_seconds", "Memory save latency", ["backend"])
MEMORY_ERRORS = Counter("memory_errors_total", "Memory errors", ["backend", "error_type"])

async def load_with_metrics(self, inputs: dict) -> dict:
    start = time.perf_counter()
    try:
        result = await self._load_memory_variables(inputs)
        MEMORY_LOAD_LATENCY.labels(backend=self.backend).observe(time.perf_counter() - start)
        return result
    except Exception as e:
        MEMORY_ERRORS.labels(backend=self.backend, error_type=type(e).__name__).inc()
        raise

Alert on p95 load latency > 50ms (Redis) or > 200ms (Postgres). Alert on error rate > 0.1%.

Common pitfalls checklist

Production hardening

Graceful degradation: If the memory store is unavailable, fall back to in-memory buffer for the request duration and log a warning. Don’t fail the user’s request because Redis hiccuped.

async def get_memory_for_session(session_id: str) -> BaseMemory:
    try:
        return await RedisMemory.create(session_id, redis_pool)
    except redis.RedisError:
        logger.warning("Redis unavailable, using ephemeral memory", session_id=session_id)
        return ConversationBufferWindowMemory(k=5, return_messages=True)

Migration path: Version your memory schema. Add a schema_version field. When you change message format (e.g., adding tool calls), write a migration script and bump the version. Read logic handles both versions.

def deserialize_messages(data: dict, version: int) -> list[BaseMessage]:
    if version == 1:
        return messages_from_dict(data["messages"])
    elif version == 2:
        return messages_from_dict_v2(data["messages"])
    else:
        raise ValueError(f"Unknown schema version: {version}")

Testing: Load-test memory in isolation. Simulate 1000 concurrent sessions, 10 turns each, with 10% concurrent writes per session. Verify no lost updates, p99 latency < 100ms, memory usage stable.

Closing thought

Multi-user memory is not a LangChain problem — it’s a distributed systems problem. The framework gives you primitives; you supply the isolation, persistence, concurrency control, and observability. Treat memory like any other critical datastore: version it, monitor it, load-test it, and plan for its failure modes. The code above is a starting point, not a finish line.