I Spent 3 Hours Debugging a LangChain Memory Bug — Here's How I Fixed It with Automated Tests

At 1:47 AM, PagerDuty woke me up. Checking the logs, I saw the agent had merged the user’s three-day-old order address with the new one, generating a nonexistent shipping address. My instinct: memory mix-up.

Such "ghost data" scenarios weren’t new in our customer service agent built with LangChain. The user would be chatting smoothly, and suddenly the context would disappear or get polluted with history from another conversation. The business team was pressing hard, so we decided to write a comprehensive automated test suite to enforce memory reliability — and ended up hitting way more pitfalls than expected.

Problem Breakdown

LangChain's Memory components (ConversationBufferMemory, ConversationSummaryMemory, etc.) are essentially abstractions over conversation history, but by default they store the history list in a plain Python list in memory. That works fine in development, but in production, three fatal issues emerge:

1. Non-persistent: When the process restarts or the Pod gets rescheduled, all memory just vanishes.
2. No isolation guarantee: Session IDs are entirely managed by the business logic, and it's easy to pull up the wrong history.
3. Serialization traps: After switching the storage backend, loading HumanMessage/AIMessage objects often gets corrupted or throws exceptions outright.

Typical unit tests only mock the Memory interface — essentially verifying "I called save, and yes, it called save" — but never check whether the saved content can be correctly restored. That’s why memory occasionally "drops frames" in production: the tests never touched the real I/O path.

Solution Design

To validate reliability, the tests must hit a real storage backend and cover three core scenarios:

• Write/Read Consistency: Multi-turn conversations stored should come back with identical order, roles, and content.
• Recovery after Restart: Simulate a process restart, reinitialize the Memory, and ensure all history is loaded intact.
• Session Isolation: No cross-talk between different sessions.

On the tech stack, we chose pytest for the test framework (fixtures make managing storage resources so easy), and used local SQLiteChatMessageHistory as the storage backend, which swaps seamlessly with Redis in production. Why not unittest? Because parameterized tests and fixture scoping let a single suite cover all memory types with way less maintenance. Why not just a mock layer? Because serialization/deserialization pitfalls lie entirely on the I/O path, and mocks can never catch them.

Core Implementation

1. Fixed Session Scenario: Verifying Read/Write Consistency

This code verifies that you get out exactly what you put into a conversation. It uses ConversationBufferMemory backed by real SQLite storage, writes three turns of dialog, then reloads from the same session and asserts each message’s type and content match exactly.

import pytest
from langchain.memory import ConversationBufferMemory
from langchain.memory.chat_message_histories import SQLiteChatMessageHistory
from langchain.schema import HumanMessage, AIMessage

SESSION_ID = "test_session_01"
DB_PATH = ":memory:"   # 生产换成文件路径即可

@pytest.fixture
def memory():
    # 每个测试独立的历史记录实例，避免污染
    history = SQLiteChatMessageHistory(session_id=SESSION_ID, connection_string=f"sqlite:///{DB_PATH}")
    mem = ConversationBufferMemory(chat_memory=history, return_messages=True)
    mem.reset()  # 清空可能残留的数据
    return mem

def test_save_and_load_complete_dialog(memory):
    # 模拟三回合对话
    dialog = [
        HumanMessage(content="我想查一下上周三的订单"),
        AIMessage(content="好的，请提供您的下单手机号"),
        HumanMessage(content="13800138000"),
        AIMessage(content="已找到，订单尾号0921，收货地址是北京朝阳区XX路"),
        HumanMessage(content="帮我改地址到上海浦东新区YY路"),
        AIMessage(content="已更新，新地址将在2小时后生效"),
    ]

    for msg in dialog:
        # save_context要求成对的input/output，我们拆开按顺序存
        if isinstance(msg, HumanMessage):
            memory.chat_memory.add_user_message(msg.content)
        else:
            memory.chat_memory.add_ai_message(msg.content)

    # 重新加载历史
    loaded = memory.load_memory_variables({})["history"]

    assert len(loaded) == len(dialog), f"消息数量不一致: 期望{len(dialog)}, 实际{len(loaded)}"
    for i, (orig, loaded_msg) in enumerate(zip(dialog, loaded)):
        assert type(orig) == type(loaded_msg), f"第{i}条消息类型不匹配"
        assert orig.content == loaded_msg.content, f"第{i}条消息内容不一致"

2. Simulating a Restart: Testing Persistent Recovery

This section answers the question: does memory survive a service restart? We create a brand-new Memory object pointing to the same session ID and verify that the history hasn’t been wiped.

def test_test_restart_recovery():
    session = "restart_test"
    db_url = "sqlite:///test_memory.db"

    # 第一次写入
    hist1 = SQLiteChatMessageHistory(session_id=session, connection_string=db_url)
    mem1 = Conversati