Migrating LangChain memory to the message history API

LangChain’s legacy memory classes (ConversationBufferMemory, ConversationSummaryMemory, and their variants) are deprecated as of v0.2. The replacement is the BaseChatMessageHistory abstraction paired with RunnableWithMessageHistory. This migration isn’t a drop-in swap — it changes how you inject history into chains, how you persist conversations, and how you test. Below is the exact sequence I use to move production workloads over without losing context or breaking streaming.

Step 1: Audit your current memory usage

Before touching code, grep for every memory import and instantiation. You’re looking for three patterns:

# Find all memory imports
grep -r "from langchain.memory" --include="*.py" .

# Find instantiations
grep -r "ConversationBufferMemory\|ConversationSummaryMemory\|ConversationBufferWindowMemory" --include="*.py" .

# Find where memory gets attached to chains
grep -r "memory=" --include="*.py" .

Typical legacy code looks like this:

# legacy_chain.py
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
memory = ConversationBufferMemory(return_messages=True)
chain = ConversationChain(llm=llm, memory=memory, verbose=True)

response = chain.invoke({"input": "Hi, I'm building a RAG system."})

Note the return_messages=True flag — that’s the hint that you’re already working with BaseMessage objects, which makes migration cleaner. If you see return_messages=False (the old default), you’re dealing with raw strings and will need a conversion step.

Step 2: Choose your message history implementation

BaseChatMessageHistory is an interface. You need a concrete backend. The ecosystem provides several; pick one that matches your persistence requirements:

For this guide I’ll use InMemoryChatMessageHistory for clarity, then show the Redis swap. Install the core package if you haven’t:

pip install -U langchain-core langchain-openai

Step 3: Replace the memory object with a history factory

The new pattern uses a session-aware factory function that returns a BaseChatMessageHistory instance. This function receives a session_id and optionally a user_id — you control the keying strategy.

# history_factory.py
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage
from langchain_community.chat_message_histories import InMemoryChatMessageHistory

# In-memory store for demo. Replace with Redis/Postgres in prod.
_store: dict[str, InMemoryChatMessageHistory] = {}

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

For Redis, the factory becomes:

# history_factory_redis.py
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_redis import RedisChatMessageHistory
import os

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

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    return RedisChatMessageHistory(
        session_id=session_id,
        url=_redis_url,
        ttl=86400,  # 24h expiry
    )

The factory pattern is deliberate: it lets you scope history per user, per conversation, or per arbitrary key without changing chain code.

Step 4: Wrap your chain with RunnableWithMessageHistory

RunnableWithMessageHistory takes your core runnable (the LLM + prompt), the history factory, and configuration keys that map input/output fields to history messages.

# migrated_chain.py
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from history_factory import get_session_history

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

with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

Key parameters:

• input_messages_key: the key in your invocation dict that holds the new user message. Must match the prompt’s human message variable ({input} above).
• history_messages_key: the variable name the prompt expects for prior messages (MessagesPlaceholder(variable_name="history")).

Step 5: Invoke with a session_id config

Every call now requires a config dict containing configurable.session_id. This is how the factory knows which history to load.

# run_migrated.py
from migrated_chain import with_history

config = {"configurable": {"session_id": "user-123-conv-456"}}

# First turn
resp1 = with_history.invoke({"input": "Hi, I'm building a RAG system."}, config=config)
print(resp1.content)

# Second turn — history automatically included
resp2 = with_history.invoke({"input": "What vector DB would you recommend?"}, config=config)
print(resp2.content)

Output shows the model referencing the first message. No manual memory.load_memory_variables() or memory.save_context() calls.

Step 6: Migrate streaming code

If you streamed with chain.stream() or chain.astream(), the wrapper preserves streaming. The only change is passing config.

# streaming_example.py
from migrated_chain import with_history

config = {"configurable": {"session_id": "stream-demo-001"}}

for chunk in with_history.stream({"input": "Explain RAG in three sentences."}, config=config):
    print(chunk.content, end="", flush=True)
print()

astream() and astream_events() work identically. The history is appended after the full response completes, so intermediate chunks don’t see the new message — same behavior as the old ConversationChain.

Step 7: Handle the return_messages=False migration

If your legacy code used ConversationBufferMemory(return_messages=False), the memory stored a single string blob. The new API expects a list of BaseMessage objects. You have two options:

Option A: Convert on load (one-time migration)

# migrate_legacy.py
from langchain_core.messages import HumanMessage, AIMessage
from langchain_community.chat_message_histories import InMemoryChatMessageHistory

def migrate_string_history(legacy_buffer: str) -> InMemoryChatMessageHistory:
    """
    Legacy buffer format:
    Human: Hello
    AI: Hi there
    Human: How are you?
    AI: I'm doing well
    """
    history = InMemoryChatMessageHistory()
    lines = legacy_buffer.strip().split("\n")
    for line in lines:
        if line.startswith("Human: "):
            history.add_message(HumanMessage(content=line[len("Human: "):]))
        elif line.startswith("AI: "):
            history.add_message(AIMessage(content=line[len("AI: "):]))
    return history

Run this once per session at startup, then save the resulting InMemoryChatMessageHistory to your persistent backend.

Option B: Custom history class that parses strings

Only worth it if you have millions of existing string buffers and can’t run a migration job. Subclass BaseChatMessageHistory and implement messages property + add_message/clear. Not shown here — prefer Option A.

Step 8: Migrate ConversationSummaryMemory

ConversationSummaryMemory periodically summarizes old turns. The new equivalent is not a different history class — it’s a separate summarization chain you run as a background job or on a schedule, then truncate the underlying history.

# summarizer.py
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.chat_history import BaseChatMessageHistory

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

SUMMARIZATION_PROMPT = """Summarize the following conversation in 3-4 sentences.
Focus on key facts, decisions, and open questions.

Conversation:
{history}

Summary:"""

async def summarize_and_truncate(
    history: BaseChatMessageHistory,
    max_messages: int = 20,
    keep_recent: int = 6,
) -> None:
    messages = history.messages
    if len(messages) <= max_messages:
        return

    # Messages to summarize (all but the most recent keep_recent)
    to_summarize = messages[:-keep_recent]
    recent = messages[-keep_recent:]

    history_text = "\n".join(
        f"{'Human' if isinstance(m, HumanMessage) else 'AI'}: {m.content}"
        for m in to_summarize
    )

    summary = await llm.ainvoke(SUMMARIZATION_PROMPT.format(history=history_text))

    # Replace history with summary + recent messages
    history.clear()
    history.add_message(SystemMessage(content=f"Conversation summary: {summary.content}"))
    for m in recent:
        history.add_message(m)

Call this from a background worker, a FastAPI lifespan hook, or after every N turns. The history backend stays the same — you’re just mutating its contents.

Step 9: Update tests

Your tests likely instantiate memory directly. Swap to the factory pattern.

# test_migrated_chain.py
import pytest
from langchain_core.messages import HumanMessage, AIMessage
from migrated_chain import with_history
from history_factory import get_session_history, _store

@pytest.fixture(autouse=True)
def clear_store():
    _store.clear()
    yield
    _store.clear()

def test_conversation_persists_across_turns():
    config = {"configurable": {"session_id": "test-session"}}

    # Turn 1
    resp1 = with_history.invoke({"input": "My name is Alex."}, config=config)
    assert "Alex" in resp1.content

    # Turn 2 — new invocation, same session
    resp2 = with_history.invoke({"input": "What's my name?"}, config=config)
    assert "Alex" in resp2.content

    # Verify history was saved
    history = get_session_history("test-session")
    assert len(history.messages) == 4  # Human, AI, Human, AI
    assert isinstance(history.messages[0], HumanMessage)
    assert history.messages[0].content == "My name is Alex."

For integration tests with Redis, spin up a test container via testcontainers-python or use fakeredis.

Step 10: Verify in production with a canary

Deploy the migrated code behind a feature flag or canary route. Compare these metrics against the legacy path:

A minimal verification script you can run against a live endpoint:

# verify_canary.py
import os
import httpx
from migrated_chain import with_history

BASE_URL = os.getenv("CANARY_URL", "http://localhost:8000")
SESSION_ID = "canary-verify-001"

def test_via_api():
    """If you exposed the chain via FastAPI/Starlette."""
    with httpx.Client(base_url=BASE_URL, timeout=30) as client:
        r1 = client.post("/chat", json={"input": "Canary test: remember the word 'pineapple'.", "session_id": SESSION_ID})
        r1.raise_for_status()
        r2 = client.post("/chat", json={"input": "What word did I ask you to remember?", "session_id": SESSION_ID})
        r2.raise_for_status()
        assert "pineapple" in r2.json()["response"].lower()
        print("API canary: PASS")

def test_direct():
    """Direct invocation (bypasses HTTP)."""
    config = {"configurable": {"session_id": SESSION_ID}}
    resp1 = with_history.invoke({"input": "Canary test: remember the word 'mango'."}, config=config)
    resp2 = with_history.invoke({"input": "What word did I ask you to remember?"}, config=config)
    assert "mango" in resp2.content.lower()
    print("Direct canary: PASS")

if __name__ == "__main__":
    test_direct()
    test_via_api()

Run this in your CI/CD post-deploy step. If it passes, promote the canary.

Common pitfalls

Missing configurable.session_id — The wrapper raises KeyError: 'session_id' with a cryptic trace. Always validate config at your API boundary:

# api_guard.py
from fastapi import HTTPException, Request

def extract_session_id(request: Request) -> str:
    session_id = request.headers.get("X-Session-ID") or request.query_params.get("session_id")
    if not session_id:
        raise HTTPException(400, "Missing session_id header or query param")
    return session_id

Prompt variable mismatch — If input_messages_key doesn’t match the prompt’s human variable, the new message never reaches the model. Double-check:

# Correct: both are "input"
prompt = ChatPromptTemplate.from_messages([..., ("human", "{input}")])
with_history = RunnableWithMessageHistory(..., input_messages_key="input", ...)

# Wrong: prompt uses {question} but key is "input"
prompt = ChatPromptTemplate.from_messages([..., ("human", "{question}")])
with_history = RunnableWithMessageHistory(..., input_messages_key="input", ...)  # BUG

History not persisting — You instantiated InMemoryChatMessageHistory inside the factory but forgot to return the same instance per session. The factory must be stateful (dict, Redis connection pool, etc.).

Streaming drops history — If you call stream() but never await the final result (or break early), the callback that saves history may not fire. Always consume the full stream or use invoke/ainvoke for fire-and-forget.

Cleanup: remove legacy imports

Once the canary is promoted and verified, delete:

# Remove from requirements.txt / pyproject.toml
# langchain (the meta-package) may still be needed for other modules
# but you can drop explicit memory imports from code

grep -r "from langchain.memory" --include="*.py" .  # should return 1

Run your full test suite. Green means done.

The migration is mechanical but touches every conversational surface in your codebase. The factory pattern pays off when you later need per-tenant isolation, audit logs, or a backend swap — changes that would have required rewriting every chain under the old memory= API.