LangChain memory with Redis for multi-session chatbots

LangChain memory with Redis solves the core problem of multi-session chatbots: keeping conversation history isolated per user while surviving process restarts and horizontal scaling. In-memory buffers vanish when your worker dies; Redis persists them with sub-millisecond latency. This guide walks through a complete implementation you can drop into a FastAPI or Flask service today.

Step 1: Provision Redis and install dependencies

Start with a Redis instance. For local development, Docker is fastest:

docker run -d --name redis -p 6379:6379 redis:7-alpine

In production, use a managed service (AWS ElastiCache, Azure Cache for Redis, or Upstash for serverless). Enable TLS and AUTH.

Install the Python packages. Pin versions to avoid surprise upgrades:

pip install "langchain==0.2.16" "langchain-openai==0.1.25" "langchain-community==0.2.16" "redis==5.0.8" "python-dotenv==1.0.1"

Verify the connection works before writing application code:

# test_redis.py
import redis
import os

r = redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379/0"))
r.ping()
print("Redis reachable")

Run it. You should see Redis reachable. If not, check firewall rules and the REDIS_URL format.

Step 2: Choose the right LangChain memory class

LangChain ships two Redis-backed memory classes. Pick one based on your access pattern:

For most multi-session chatbots, the second option is cleaner. It returns a history key compatible with MessagesPlaceholder in prompts.

# memory_factory.py
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain.memory import ConversationBufferWindowMemory

def get_memory(session_id: str, window_size: int = 10) -> ConversationBufferWindowMemory:
    history = RedisChatMessageHistory(
        session_id=session_id,
        url=os.getenv("REDIS_URL", "redis://localhost:6379/0"),
        key_prefix="chatbot:",  # namespaces keys: "chatbot:{session_id}"
        ttl=60 * 60 * 24 * 30,  # 30 days; adjust for your retention policy
    )
    return ConversationBufferWindowMemory(
        chat_memory=history,
        k=window_size,
        return_messages=True,  # critical: returns BaseMessage list, not string
        memory_key="history",
        input_key="input",
        output_key="output",
    )

The key_prefix prevents collisions if you share a Redis DB with other services. The ttl auto-expires abandoned sessions — no cleanup cron needed.

Step 3: Wire memory into a chain

Use RunnableWithMessageHistory (the LCEL way) so the same chain works for streaming and non-streaming calls. This pattern also makes it trivial to swap the model later.

# chain.py
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from memory_factory import get_memory

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise assistant. Answer in 2-3 sentences."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
chain = prompt | llm

# Wrap with message history
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_memory,
    input_messages_key="input",
    history_messages_key="history",
)

Notice get_memory is a callable that accepts session_id — RunnableWithMessageHistory invokes it per request. This is the hook that makes multi-session work.

Step 4: Expose a session-aware HTTP endpoint

FastAPI example. The session ID comes from a header, cookie, or JWT claim — never from the request body (that lets users hijack sessions).

# main.py
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
from chain import chain_with_history
import uuid

app = FastAPI()

class ChatRequest(BaseModel):
    input: str

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

@app.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest,
    x_session_id: str | None = Header(default=None, alias="X-Session-ID"),
):
    session_id = x_session_id or str(uuid.uuid4())
    config = {"configurable": {"session_id": session_id}}
    
    result = await chain_with_history.ainvoke(
        {"input": request.input},
        config=config,
    )
    
    return ChatResponse(output=result.content, session_id=session_id)

Run it:

uvicorn main:app --reload --port 8000

Step 5: Verify multi-session isolation

Open two terminals. Each simulates a different user.

Terminal A — User Alice:

SESSION_ID=$(uuidgen)
curl -s -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -H "X-Session-ID: $SESSION_ID" \
  -d '{"input": "My favorite color is teal."}' | jq

Terminal B — User Bob:

curl -s -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -H "X-Session-ID: $SESSION_ID" \
  -d '{"input": "What is my favorite color?"}' | jq

Alice gets a response acknowledging teal. Bob gets a generic answer — he has no history. Swap the session IDs and the behavior follows the ID, not the connection.

Inspect Redis directly

Confirm keys exist and have the right shape:

redis-cli --scan --pattern "chatbot:*"
# chatbot:550e8400-e29b-41d4-a716-446655440000

redis-cli GET "chatbot:550e8400-e29b-41d4-a716-446655440000"
# Returns JSON array of messages with type, content, additional_kwargs

Each session is a separate key. TTL counts down. Restart the FastAPI process — history survives.

Step 6: Handle streaming responses

Production chatbots stream tokens. RunnableWithMessageHistory supports astream and astream_log out of the box. The memory only updates after the full response completes, so partial tokens don’t pollute history.

# streaming_endpoint.py
from fastapi.responses import StreamingResponse
import json

@app.post("/chat/stream")
async def chat_stream(
    request: ChatRequest,
    x_session_id: str | None = Header(default=None, alias="X-Session-ID"),
):
    session_id = x_session_id or str(uuid.uuid4())
    config = {"configurable": {"session_id": session_id}}
    
    async def token_generator():
        async for chunk in chain_with_history.astream(
            {"input": request.input},
            config=config,
        ):
            yield f"data: {json.dumps({'token': chunk.content})}\n\n"
        yield f"data: {json.dumps({'session_id': session_id, 'done': True})}\n\n"
    
    return StreamingResponse(token_generator(), media_type="text/event-stream")

Test with curl -N (no buffer):

curl -N -X POST http://localhost:8000/chat/stream \
  -H "Content-Type: application/json" \
  -H "X-Session-ID: $SESSION_ID" \
  -d '{"input": "Count to five."}'

You’ll see tokens arrive incrementally, then a final done event with the session ID.

Step 7: Trim aggressively to control context window and cost

ConversationBufferWindowMemory keeps the last k message pairs (human + AI). That’s often too much for long conversations. Add a summarization step when the buffer exceeds a threshold.

# memory_factory.py (extended)
from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI

def get_memory(session_id: str, window_size: int = 10, max_tokens: int = 2000) -> ConversationBufferWindowMemory | ConversationSummaryBufferMemory:
    history = RedisChatMessageHistory(
        session_id=session_id,
        url=os.getenv("REDIS_URL", "redis://localhost:6379/0"),
        key_prefix="chatbot:",
        ttl=60 * 60 * 24 * 30,
    )
    
    # Use summary memory for long-running sessions
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    return ConversationSummaryBufferMemory(
        chat_memory=history,
        llm=llm,
        max_token_limit=max_tokens,
        return_messages=True,
        memory_key="history",
        input_key="input",
        output_key="output",
    )

ConversationSummaryBufferMemory keeps recent messages verbatim and summarizes older ones. The summary counts toward max_token_limit. Tune max_tokens to your model’s context window minus prompt overhead.

Step 8: Add per-session metadata (optional but useful)

Attach user ID, feature flags, or experiment buckets to the Redis hash. RedisChatMessageHistory stores messages in a list, but you can write a sidecar key.

# metadata.py
import json
import redis
import os

r = redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379/0"))

def set_session_meta(session_id: str, meta: dict):
    key = f"chatbot:meta:{session_id}"
    r.hset(key, mapping={k: json.dumps(v) for k, v in meta.items()})
    r.expire(key, 60 * 60 * 24 * 30)

def get_session_meta(session_id: str) -> dict:
    key = f"chatbot:meta:{session_id}"
    data = r.hgetall(key)
    return {k: json.loads(v) for k, v in data.items()}

Call set_session_meta when the session is created (first request or auth callback). Read it in middleware for logging, rate limiting, or A/B routing.

Step 9: Observability — log memory ops without slowing requests

Wrap the memory factory to emit structured logs. Use structlog or stdlib logging with a JSON formatter.

# memory_factory.py (add logging)
import logging
import time
from functools import wraps

logger = logging.getLogger("chatbot.memory")

def logged_memory_factory(session_id: str, **kwargs):
    start = time.perf_counter()
    mem = get_memory(session_id, **kwargs)
    duration_ms = (time.perf_counter() - start) * 1000
    logger.info(
        "memory_created",
        session_id=session_id,
        duration_ms=round(duration_ms, 2),
        memory_type=type(mem).__name__,
    )
    return mem

In your endpoint, swap get_memory for logged_memory_factory. You’ll see latency per session creation — useful for spotting cold-start Redis latency.

Step 10: Deploy with connection pooling and health checks

Don’t create a new Redis connection per request. The redis-py client pools by default when you use from_url, but verify the pool size matches your worker count.

# redis_pool.py
import redis
import os

pool = redis.ConnectionPool.from_url(
    os.getenv("REDIS_URL", "redis://localhost:6379/0"),
    max_connections=50,  # tune to worker_concurrency * 2
    decode_responses=True,
    socket_keepalive=True,
    socket_connect_timeout=5,
    socket_timeout=5,
    retry_on_timeout=True,
    health_check_interval=30,
)

def get_redis_client() -> redis.Redis:
    return redis.Redis(connection_pool=pool)

Pass this client to RedisChatMessageHistory via the redis_client parameter instead of url:

history = RedisChatMessageHistory(
    session_id=session_id,
    redis_client=get_redis_client(),
    key_prefix="chatbot:",
    ttl=60 * 60 * 24 * 30,
)

Add a /healthz endpoint that pings Redis:

@app.get("/healthz")
async def healthz():
    try:
        get_redis_client().ping()
        return {"status": "ok", "redis": "connected"}
    except Exception as e:
        raise HTTPException(503, detail=f"Redis unavailable: {e}")

Load balancers and Kubernetes liveness probes need this.

Step 11: Test failure modes locally

Simulate Redis outage to verify graceful degradation:

docker pause redis
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -H "X-Session-ID: test" \
  -d '{"input": "hello"}'
# Expect 503 from /healthz, 500 from /chat (or your custom error handler)
docker unpause redis

Decide your policy: return a friendly error, fall back to in-memory buffer for the request, or queue the message. The key is deciding explicitly rather than letting a stack trace leak to the user.

Step 12: Migrate existing in-memory sessions (if applicable)

If you’re moving from ConversationBufferMemory in a single-process app, write a one-off script to backfill Redis:

# migrate.py
import json
from memory_factory import get_memory

# Assume you have a pickle file or JSON dump of old sessions
with open("old_sessions.json") as f:
    old = json.load(f)

for session_id, messages in old.items():
    memory = get_memory(session_id)
    for msg in messages:
        if msg["type"] == "human":
            memory.chat_memory.add_user_message(msg["content"])
        elif msg["type"] == "ai":
            memory.chat_memory.add_ai_message(msg["content"])
    print(f"Migrated {session_id}: {len(messages)} messages")

Run once, verify key counts in Redis, then delete the script.

Verification checklist

• Two different X-Session-ID values maintain independent histories
• Restarting the API process does not lose history
• redis-cli --scan --pattern "chatbot:*" shows one key per session
• TTL decreases over time; expired keys vanish automatically
• Streaming endpoint yields tokens, then a final done event
• /healthz returns 200 when Redis is up, 503 when paused
• Memory creation latency logs stay under 50 ms p99
• Summarization kicks in for sessions exceeding max_token_limit

Common pitfalls

This pattern scales horizontally: add more API replicas, they all share the same Redis. The only stateful component is Redis itself, which you already operate as a managed service. When you eventually need per-user personalization, RAG context, or tool-use history, the same session_id key becomes the anchor for all of it.