前言

多轮对话需要记忆。RAGFlow 虽然有原生会话功能，但本项目选择了 LangChain + MongoDB 的方案来独立管理对话记忆。这样做的好处是：RAGFlow 负责检索，LangChain 负责记忆，职责分离，各自使用最擅长的工具。本文介绍基于 ConversationBufferMemory + MongoDBChatMessageHistory 的持久化方案。

为什么不用 RAGFlow 原生会话？

RAGFlow 的会话是绑定在特定 Assistant 上的。项目采用临时会话模式（创建→提问→删除），如果还要用 RAGFlow 管理对话记忆，就会出现两个问题：

1. 不能删除会话（因为删了就丢了历史）

2. 切换不同助手回答时，历史分散在不同会话中

因此选择在 LangChain 侧统一管理对话记忆。

数据流向

用户提问
    ↓
从 MongoDB 加载该 session 的历史消息
    ↓
格式化为对话文本 → 注入 Prompt
    ↓
LLM 生成回答
    ↓
新的一问一答存入 MongoDB

核心实现

创建记忆对象

# Langchain_utils/memory.py
​
from langchain.memory import ConversationBufferMemory
from langchain_mongodb import MongoDBChatMessageHistory
​
def create_memory(session_id: str = "default") -> ConversationBufferMemory:
    connection_string = f"{MONGODB_URI.rstrip('/')}/{MONGODB_DB}"
​
    # MongoDB 消息历史存储
    message_history = MongoDBChatMessageHistory(
        connection_string=connection_string,
        collection_name=MONGODB_COLLECTION,  # 如 "chat_history"
        database_name=MONGODB_DB,            # 如 "conversation_db"
        session_id=session_id                 # 区分不同会话
    )
​
    # 缓冲记忆（保留所有历史）
    memory = ConversationBufferMemory(
        memory_key='chat_history',
        chat_memory=message_history,
        input_key='question',
        output_key='answer',
        return_messages=True
    )
​
    return memory

MongoDBChatMessageHistory 是 LangChain 官方提供的 MongoDB 集成，底层封装了 pymongo。每个 session_id 对应 MongoDB 中的一组文档。

MongoDB 中的存储结构

每个对话消息在 MongoDB 中以如下形式存在：

{
    "_id": ObjectId("..."),
    "SessionId": "user_12345",
    "History": {
        "type": "human",
        "data": {
            "content": "公司的请假流程是什么？",
            "additional_kwargs": {}
        }
    }
}

关键字段：

• SessionId：对应 session_id，用于隔离不同用户/会话

• History.type："human" 或 "ai"

• History.data.content：消息文本

格式化对话历史

从 MongoDB 取出的消息是 LangChain 的消息对象列表，需要格式化为 LLM 可读的文本：

def format_chat_history(messages: List[Any]) -> str:
    formatted = ""
    for msg in messages:
        if isinstance(msg, HumanMessage):
            formatted += f"用户: {msg.content}\n"
        elif isinstance(msg, AIMessage):
            formatted += f"助手: {msg.content}\n"
    return formatted

格式化后的文本示例：

用户: 你好，请问你是谁？
助手: 你好！我是企业知识库助手，可以回答公司相关问题。
用户: 公司的请假流程是什么？

这段文本最终被注入到 LLM 的 Prompt 中作为对话上下文。

在处理器中的读写时机

# 读取 —— 在处理问题之前
memory_variables = memory.load_memory_variables({})
chat_history = memory_variables.get("chat_history", [])
​
# 写入 —— 在生成回答之后
memory.chat_memory.add_messages([
    HumanMessage(content=question),     # 用户问题
    AIMessage(content=full_answer)      # 模型回答
])

写入用的是 add_messages，会追加到 MongoDB 中。load_memory_variables({"chat_history"}) 返回 {"chat_history": [HumanMessage, AIMessage, ...]}。

会话隔离

不同用户/浏览器对应不同的 session_id：

# output/main_service.py
​
_processors: Dict[str, Any] = {}
​
async def get_response(question: str, session_id: str):
    if session_id not in _processors:
        # 为新会话创建 processor（内部会 create_memory(session_id)）
        _processors[session_id] = create_unified_processor(session_id=session_id)
​
    processor = _processors[session_id]
    async for chunk in processor(question):
        yield chunk

每个 session 有独立的：

• Processor 实例

• ConversationBufferMemory（对应 MongoDB 中不同 SessionId 的文档）

• 意图历史

清除记忆也是按 session_id 操作：

def clear_session_memory(session_id: str):
    if session_id in _processors:
        processor = _processors[session_id]
        processor.clear_memory()  # 内部调用 memory.clear()

ConversationBufferMemory 的局限性

ConversationBufferMemory 保留所有历史消息。这在对话很长时会带来两个问题：

1. Token 消耗越来越大：每次请求都带着完整历史

2. MongoDB 文档不断增长：虽然慢但持续增长

项目中当前没有做窗口截断，生产环境可以考虑：

• ConversationBufferWindowMemory：只保留最近 N 轮对话

• ConversationSummaryMemory：用 LLM 对历史做摘要

配置管理

# Langchain_utils/config.py
​
MONGODB_URI = os.getenv("", "")
MONGODB_DB = os.getenv("", "")
MONGODB_COLLECTION = os.getenv("", "")
DEFAULT_SESSION_ID = ""
SESSION_TTL_DAYS =   # 预留的过期时间配置

通过环境变量覆盖默认值，方便不同环境切换。

异常处理

MongoDB 不可用时不应该让整个对话挂掉：

try:
    memory.chat_memory.add_messages([input_message, output_message])
except Exception as e:
    logging.error(f"将会话历史保存到MongoDB时出错: {e}")

保存失败只记日志，不影响当前对话。代价是这条消息丢失了，下次对话时缺少这一轮上下文，但用户体验不受影响。

小结

本文介绍了基于 LangChain + MongoDB 的对话记忆方案：

1. MongoDBChatMessageHistory：开箱即用的 MongoDB 持久化

2. ConversationBufferMemory：全量保留历史的缓冲策略

3. Session 隔离：每个 session_id 独立存储，互不干扰

4. 异常降级：MongoDB 不可用时对话不中断