我花了6小时修复 LangChain 的 ConversationBufferMemory —— 这是你需要的自动化测试

Source: Dev.to

背景

在星期五下午 4:59 ，QA 同事报告说支持机器人记住的用户是 Zhang San，但在询问订单号时，它的回复却像是针对用户 Li Si。日志显示 LangChain 的 ConversationBufferMemory 在会话之间混合了聊天历史。于是我们意识到，需要一个自动化测试套件来在下一次深夜事故发生之前，锁定记忆存储的准确性和一致性。

LLM 驱动聊天产品的挑战

• 记忆模块必须在多轮对话中保留上下文（例如，“我住在北京” → 后续的天气查询）。
• ConversationBufferMemory 将对话存储为纯文本，这在数据能够放入 RAM 时可行，但持久化到 Redis 或数据库时会出现问题。
• 持久化会引入序列化/反序列化问题、并发读写以及旧消息的裁剪。
• 手动 QA 可能遗漏竞争条件和边缘情况，例如 trim_messages 在 Redis 连接中断时会混淆相邻会话。

在生产环境中，某客服机器人同时服务数百名用户，所有用户共享同一个 Redis 实例。手动测试未发现跨会话泄漏，但真实流量很快暴露出像打地鼠一样出现又消失的 bug。

解决方案概述

目标是在 CI 中 运行核心记忆逻辑，而不需要真实的 LLM 或 Redis 实例，以在代码合并前捕获回归。

• 测试框架： pytest – 其 fixture 系统能够干净地组装不同的 memory 实例。
• Redis 模拟： fakeredis – 一个内存中的 mock Redis，没有任何副作用。
• LLM 调用： 使用 unittest.mock 进行 mock，因为重点是 memory，而不是语言生成。

内置的 langchain.tests 只覆盖了浅层接口，遗漏了诸如消息类型转换和多会话隔离等场景，因此需要自定义测试套件。运行真实的 Redis 容器会给 CI 构建增加约 3 分钟，这是不可接受的。

架构

1. 在 conftest.py 中定义 fake_redis_memory fixture。
2. 使用该 fixture 构造各种 Memory 子类（ConversationBufferMemory、ConversationSummaryMemory）。
3. 使用辅助函数模拟多轮对话。
4. 断言 load_memory_variables 返回完整且会话隔离的历史记录。

所有测试必须 不进行任何网络请求，且每个测试的执行时间不超过 0.3 秒。

Test Fixture (conftest.py)

# conftest.py
import pytest
from unittest.mock import MagicMock
from langchain.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import RedisChatMessageHistory
from fakeredis import FakeRedis

@pytest.fixture
def fake_redis_memory():
    """创建一个工厂，返回一个基于虚假 Redis 的 ConversationBufferMemory。"""
    fake_redis_client = FakeRedis()

    def _create_memory(session_id: str):
        # 注入虚假 Redis 客户端以确保会话隔离。
        history = RedisChatMessageHistory(
            session_id=session_id,
            redis_client=fake_redis_client
        )
        # `return_messages=True` 会返回 Message 对象，便于断言。
        memory = ConversationBufferMemory(
            chat_memory=history,
            return_messages=True
        )
        return memory

    return _create_memory

测试用例 (test_memory_accuracy.py)

# test_memory_accuracy.py
from langchain.schema import HumanMessage, AIMessage

def test_buffer_memory_keeps_all_messages(fake_redis_memory):
    memory = fake_redis_memory("session_1202")

    # First turn
    memory.save_context(
        {"input": "我叫张三"},
        {"output": "你好张三"}
    )
    # Second turn
    memory.save_context(
        {"input": "我的订单号是多少"},
        {"output": "你的订单号是 #1123"}
    )

    variables = memory.load_memory_variables({})
    history = variables.get("history", [])

    # Expect 4 messages: two human inputs and two AI responses.
    assert len(history) == 4
    assert isinstance(history[0], HumanMessage)
    assert isinstance(history[1], AIMessage)
    assert isinstance(history[2], HumanMessage)
    assert isinstance(history[3], AIMessage)
    assert history[0].content == "我叫张三"
    assert history[1].content == "你好张三"
    assert history[2].content == "我的订单号是多少"
    assert history[3].content == "你的订单号是 #1123"

此测试验证 所有消息都能被正确存储和检索，从而消除“我只存了两行，却只返回了一行”的 bug。

结论

通过结合 pytest、fakeredis 和模拟 LLM，我们构建了一个快速、可靠且适合 CI 的测试套件，该套件能够：

• 检测跨会话的污染。
• 验证序列化/反序列化逻辑。
• 确保修剪和持久化行为符合预期。

该套件每个测试的运行时间不足三分之一秒，无需外部服务，并且能够让我们确信 ConversationBufferMemory 在生产负载下能够正常工作。