【AI实战日记-手搓情感聊天机器人】Day 6：给 AI 装上“记忆体”！使用 Redis 实现对话记忆持久化

前几天的实战中，我们的对话记录都保存在内存（RAM）中，程序重启即丢失。今天是 Day 6，我们将引入工业级缓存数据库 Redis，配合 LangChain Community 的 RedisChatMessageHistory 组件，彻底解决记忆丢失问题。本文将手把手教你使用 Docker 快速部署 Redis，并重构代码实现 Session 级别的持久化存储，让 Project Echo 真正

qqq2199

551人浏览 · 2026-01-07 09:22:28

qqq2199 · 2026-01-07 09:22:28 发布

前几天的实战中，我们的对话记录都保存在内存（RAM）中，程序重启即丢失。今天是 Day 6，我们将引入工业级缓存数据库 Redis，配合 LangChain Community 的 RedisChatMessageHistory 组件，彻底解决记忆丢失问题。本文将手把手教你使用 Docker 快速部署 Redis，并重构代码实现 Session 级别的持久化存储，让 Project Echo 真正拥有“长期记忆”。

一、项目进度：Day 6 启动

根据 Project Echo 路线图，今天是 Phase 2 的最后一天。
搞定持久化后，我们的基础对话系统就彻底闭环了。

二、核心原理：Redis 持久化存储全流程

在引入 Redis 后，我们的对话不再是简单的内存读写，而是一个标准的 “读档 -> 生成 -> 存档” 闭环。这一过程完全由 LangChain 的 RunnableWithMessageHistory 和 RedisChatMessageHistory 组件接管。

1. 数据流转机制 (The Data Flow)

整个持久化流程基于 Session ID (会话唯一标识) 运转：

🔍 读档 (Load History)：
- 当用户发出消息时，系统首先根据 session_id（例如 "user_001"）去 Redis 数据库中查询对应的 Key。
- Redis 将存储的序列化数据（通常是 JSON 格式的历史记录）返回给系统。
- LangChain 将其反序列化为 Python 对象 (HumanMessage, AIMessage)，恢复上下文。
🧠 生成 (Generate)：
- 系统将恢复的上下文 + 当前用户输入，打包发送给 通义千问 (LLM)。
- LLM 生成回复。
💾 存档 (Save History)：
- 系统将 [当前用户输入] 和 [AI 最新回复] 追加到历史对象中。
- LangChain 将更新后的完整历史记录序列化，并写回 Redis，覆盖或追加旧数据。

2. 持久化全链路序列图 (Sequence Diagram)

这张图展示了从用户按下回车，到数据落地 Redis 的完整时序：

3. 底层存储结构揭秘

在 Redis 内部，LangChain 通常会将聊天记录存储为 List 或 JSON String。
假设你的 session_id 是 user_001，Redis 中的数据结构大致如下：

Key: message_store:user_001 (前缀可配置)

Value:

[
  {"type": "human", "data": {"content": "你好，我叫阿强"}},
  {"type": "ai", "data": {"content": "你好阿强，我是傲娇酱。"}},
  {"type": "human", "data": {"content": "我叫什么？"}},
  {"type": "ai", "data": {"content": "你叫阿强。"}}
]

正是这种结构化的存储，保证了无论程序重启多少次，只要 Key 还在，记忆就在。

三、实战：环境准备

1. 安装 Redis (推荐 Docker)

这里使用docker来安装redis，如果你装了 Docker，在终端运行这一行命令即可启动 Redis：

docker run -d --name echo-redis -p 6379:6379 redis:latest

如果没有 Docker：Windows 用户可以去 Redis 官网下载安装包，或者使用 Memurai (Redis 的 Windows 兼容版)。

2. 安装 Python 依赖

我们需要安装 redis 驱动库。

pip install redis

3. 更新配置 (src/config/settings.py)

在 .env 文件中添加 Redis 地址（通常是 redis://localhost:6379/0），并在 settings.py 中读取。

修改 src/config/settings.py:

# ... 之前的代码 ...
class Config:
    # ... 其他配置 ...
    # 新增 Redis 配置
    REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")

settings = Config()

四、实战：代码改造 (LCEL 架构)

我们需要修改 main.py，将内存存储替换为 Redis 存储。

修改文件：main.py

# ==========================================
# Day 6: Persistent Memory with Redis (LCEL)
# ==========================================

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
# 【关键修改】引入 Redis 记忆组件
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory

from src.core.llm import LLMClient
from src.core.prompts import PROMPTS
from src.utils.logger import logger
from src.core.emotion import EmotionEngine
from src.config.settings import settings # 导入配置

# --- 【关键修改】废弃全局变量 store = {} ---
# 我们不再需要在内存里存这个字典了

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    """
    工厂函数：根据 session_id 返回对应的 Redis 历史记录对象
    LangChain 会自动处理连接和读写
    """
    return RedisChatMessageHistory(
        session_id=session_id,
        url=settings.REDIS_URL, # 从配置读取连接串
        ttl=3600 * 24 * 7       # (可选) 设置记忆有效期为 7 天
    )

def main():
    logger.info("--- Project Echo: Day 6 Persistence (Redis) ---")
    
    # 1. 初始化组件 (保持不变)
    client = LLMClient()
    llm = client.get_client()
    emotion_engine = EmotionEngine()
    
    # 2. 构建 Prompt (保持不变)
    sys_prompt_base = PROMPTS["tsundere"]
    prompt = ChatPromptTemplate.from_messages([
        ("system", sys_prompt_base),
        ("system", "{emotion_context}"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}")
    ])

    # 3. 组装基础链 (保持不变)
    chain = prompt | llm

    # 4. 挂载持久化记忆 (保持不变，只是 get_session_history 变了)
    with_message_history = RunnableWithMessageHistory(
        chain,
        get_session_history, # 这里传入的是连 Redis 的函数
        input_messages_key="input",
        history_messages_key="history",
    )

    print("\n💡 Tip: 即使现在关闭程序，下次回来我也记得你！\n")
    
    # 使用固定的 Session ID 来测试持久化效果
    session_id = "user_permanent_001"

    while True:
        user_input = input("You: ")
        if user_input.lower() in ["quit", "exit"]:
            break
            
        if user_input.strip():
            # --- 情绪分析 & 策略 (保持不变) ---
            current_emotion = emotion_engine.analyze(user_input)
            emotion_instruction = "用户情绪平稳，正常交流。"
            
            if "[愤怒]" in current_emotion:
                emotion_instruction = "⚠️ 警告：用户生气了！请示弱、道歉。"
            # ... 其他情绪逻辑 ...

            try:
                # --- 执行对话 ---
                response = with_message_history.invoke(
                    {
                        "input": user_input, 
                        "emotion_context": emotion_instruction
                    },
                    config={"configurable": {"session_id": session_id}}
                )
                print(f"Bot ({current_emotion}): {response.content}\n")
                
            except Exception as e:
                logger.error(f"调用失败 (请检查Redis是否启动): {e}")

if __name__ == "__main__":
    main()