【AI】十七.大模型存储记忆实战

小滴课堂讲师 2025年09月20日 ai大模型, aigc 预计阅读 38 分钟

#### LLM大模型存储记忆功能介绍和应用场景

##### 需求背景：为什么需要存储记忆功能

* 长对话上下文遗忘问题

```markdown
  # 示例：第二次提问时模型已“失忆”
  user_input1 = "我叫张三"  
  ai_response1 = "你好张三！"  
  user_input2 = "我叫什么名字？"  
  ai_response2 = "抱歉，我不知道您的名字。"  # 期望回答“张三”
  ```

* 大模型（如 GPT-4）单次对话的上下文窗口有限（通常为 4k-128k tokens），导致多轮对话中容易丢失早期信息。
  * 例如，用户询问 “如何制作蛋糕” 后接着问 “需要烤箱吗”
  * 模型若无法记住前一轮对话，可能回答 “需要烤箱” 但忘记蛋糕配方的关键步骤。

* 个性化服务与用户偏好记忆

* 在客服、教育、医疗等场景中，用户需要模型记住个人信息（如姓名、病史）或历史行为（如订单记录、学习进度）。
  * 例如，医疗助手需根据患者的历史诊断结果提供建议。

* 复杂任务的状态管理

* 涉及多步骤的任务（如旅行规划、代码调试）需要模型跟踪中间状态。

* 例如，用户要求 “规划上海到北京的三天行程”，模型需记住已推荐的景点、交通方式等。

##### LangChain方法

* 短期记忆：通过 Memory 模块存储对话历史，确保模型在多轮交互中保持连贯

* 长期记忆：将用户数据存储在外部数据库或向量数据库（如 Milvus、Pinecone），实现跨会话的长期记忆

##### 记忆功能的核心设计

* 两种类型

* 记忆的实现方式

* **短期记忆**：通过拼接历史消息实现（`[用户: 你好][AI: 你好！]`）。
  * 长期记忆
    - **结构化存储**：用数据库记录关键信息（如用户喜好）。
    - **向量化存储**：将文本转为向量存入向量库（如Milvus）。
    - **混合模式**：短期记忆 + 长期检索增强（RAG）

##### 应用场景与案例

* 个性化教育助手 (伪代码)

```python
  from langchain.memory import VectorStoreRetrieverMemory
  # 初始化记忆系统
  retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
  memory = VectorStoreRetrieverMemory(retriever=retriever)
  # 运行示例
  memory.save_context(
      {"input": "我的学习目标是掌握微积分"}, 
      {"output": "目标已记录，将推荐相关资源"}
  )
  memory.save_context(
      {"input": "请解释洛必达法则"}, 
      {"output": f"{explanation} 已添加到你的学习清单"}
  )
  # 后续对话
  query = "根据我的目标推荐学习资料"
  relevant_memories = memory.load_memory_variables({"query": query}) 
  # 返回微积分相关记忆
  
  ```

* 电商推荐引擎

```python
  from langchain.retrievers import TimeWeightedVectorStoreRetriever
  # 带时间权重的记忆系统
  retriever = TimeWeightedVectorStoreRetriever(
      vectorstore=vectorstore,
      decay_rate=0.95,  # 记忆衰减系数
      k=5
  )
  # 记忆示例数据
  retriever.add_documents([
      Document(page_content="用户2026-12-01购买手机", metadata={"type": "purchase"}),
      Document(page_content="用户2027-03-15浏览笔记本电脑", metadata={"type": "browse"}),
      Document(page_content="用户2027-06-20退货耳机", metadata={"type": "return"})
  ])
  # 获取最新加权记忆
  relevant_memories = retriever.get_relevant_documents("用户兴趣分析")
  ```

##### 大模型长短期记忆选择决策树

* 基础用法：短期对话记忆
* 进阶用法：长期记忆 + 向量数据库
* 高级用法：多用户隔离记忆（会话级）
![1-3479667.png](https://yn-blog.oss-cn-chengdu.aliyuncs.com/agent_v1/2025-09-20/6958a461-f29c-4b57-9d47-b925cc386dd3.png)

#### LLM存储记忆功能之BaseChatMemory实战

##### BaseChatMemory 介绍

* 是 LangChain 中所有**聊天型记忆模块的基类**，定义了记忆存储和检索的通用接口。

```
  from langchain.memory.chat_memory import BaseChatMemory
  ```

* 可通过继承此类实现自定义记忆逻辑（如过滤敏感信息、动态清理策略）

* 注意：部分API虽然过期，但也需要知道核心思想, 新旧我们都有讲

* 核心作用

* **标准化接口**：统一 `save_context()`（保存上下文）和 `load_memory_variables()`（加载记忆）方法。
  * **状态管理**：维护对话历史（`chat_memory` 属性），支持消息的增删改查。
  * **扩展性**：允许开发者覆盖默认行为（如自定义存储格式、加密数据）

* 关键属性

* `chat_memory`
    * 存储对话消息的容器，类型为 ChatMessageHistory，包含 messages 列表
    * 每条消息为 BaseMessage 对象，如 HumanMessage、AIMessage

* 核心方法

* `save_context` 保存用户输入和模型输出到 chat_memory。

```
    memory.save_context({"input": "你好"}, {"output": "你好！有什么可以帮您？"})
    # 等价于：
    memory.chat_memory.add_user_message("你好")
    memory.chat_memory.add_ai_message("你好！有什么可以帮您？")
    ```

* `load_memory_variables` 返回当前记忆内容（通常为 {"history": "对话历史字符串"}

```
    variables = memory.load_memory_variables({})
    print(variables["history"])  # 输出：Human: 你好\nAI: 你好！有什么可以帮您？
    ```

* `clear()` 清空所有记忆

```
    #调用 
    chat_memory.clear()
    ```

##### BaseChatMemory的子类

##### 案例实战

* `ConversationBufferMemory`

```python
  from langchain.memory import ConversationBufferMemory
  
  # 初始化对话缓冲记忆
  memory = ConversationBufferMemory(
      memory_key="chat_history", #存储进去的Key，和获取的时候需要保持一致
      return_messages=True,
  )
  
  # 添加用户消息和 AI 消息
  memory.chat_memory.add_user_message("你的名字是什么")
  memory.chat_memory.add_ai_message("二当家")
  
  # 加载记忆内容
  memory_variables = memory.load_memory_variables({})
  print("记忆内容：", memory_variables)
  ```

* `ConversationBufferWindowMemory `

* 滑动窗口式对话记忆, 此类在对话记录中引入窗口机制，仅保存最近的 `k` 条对话历史。
  * **k 参数**：定义窗口大小，表示最多保留的对话记录条数, 适合长对话或对历史要求不高的应用。

```python
  from langchain.memory import ConversationBufferWindowMemory
  
  # 初始化窗口记忆，设置窗口大小为 2
  memory = ConversationBufferWindowMemory(k=2, memory_key="chat_history")
  
  # 保存一些消息
  memory.save_context({"input": "你叫什么名字"}, {"output": "你好，我是三当家"})
  memory.save_context({"input": "小滴课堂怎么样"}, {"output": "小滴课堂是适合初学者的计算机学习平台"})
  memory.save_context({"input": "好的，我去了解下"}, {"output": "希望对你有帮助！"})
  
  # 获取窗口内的对话历史
  window_history = memory.load_memory_variables({})
  print("当前窗口内的对话历史：", window_history)
  ```

#### 【面试题】LLM存储优化-大量长对话如何解决

##### 面试题

* 传统对话系统每次交互独立，模型无法感知历史对话内容，如何解决？
* 长对话超出模型的Token处理能力，导致信息截断或性能下降，如何解决？

##### 大模型场景题目的需求背景【重要】

* **大模型的上下文限制**
  * 大语言模型（如GPT-4、DeepSeek等虽然能处理复杂的对话任务
  * 但其输入长度存在限制（如Token上限），无法直接存储长期对话历史
* **对话连贯性需求**
  * 实际应用中（如客服系统、智能助手），用户问题常依赖上下文。
  * 例如，用户先问“人工智能的定义”，再要求“详细说明”，模型需基于历史回答才能生成合理响应
* 资源优化需求
  * 直接存储完整对话历史会占用大量内存或数据库资源，且频繁传递完整上下文会增加计算成本。

##### 面试回答要点

* **核心目标**  
  * 通过摘要存储实现对话上下文的长期维护，解决大模型Token限制与对话连贯性问题。
* 技术实现
  * 记忆模块：LangChain提供ConversationBufferMemory（完整历史）和ConversationSummaryMemory（摘要存储）等
  * 摘要生成：调用LLM对历史对话生成摘要，后续交互仅传递摘要而非完整历史
* **优势**
  - 减少Token消耗，适配模型输入限制。
  - 提升对话系统的长期记忆能力。
  - 支持分布式存储（如MongoDB、Milvus），扩展性强

##### ConversationSummaryMemory

* 通过模型生成对话的摘要，帮助保留重要信息，而不保存完整历史,适合需要长期记忆的场景。

* 原理就是提示词，让AI帮出来摘要，而且可以定制

```python
  请将以下对话压缩为简短摘要，保留用户需求和关键结果：
  对话历史：
  {history}
  
  当前对话：
  Human: {input}
  AI: {output}
  摘要：
  """
  ```

* **load_memory_variables**：返回当前会话的摘要信息。

```python
  from langchain.memory import ConversationSummaryMemory
  from langchain_openai import ChatOpenAI
  
  # 初始化大模型
  llm = ChatOpenAI(
      model_name = "qwen-plus",
      base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
      api_key="sk-005c3c25f6d042848b29d75f2f020f08",
      temperature=0.7
  )
  # 初始化摘要记忆
  memory = ConversationSummaryMemory(llm=llm)
  # 模拟对话
  memory.save_context({"input": "你叫什么名字？"}, {"output": "你好，我是三当家。"})
  memory.save_context({"input": "你能告诉我机器学习吗？"}, {"output": "机器学习是人工智能的一个分支。"})
  # 获取摘要
  summary = memory.load_memory_variables({})
  print("当前对话摘要：", summary)
  ```

#### 基于LangChain的带摘要存储对话系统实战

##### 实战案例

* 基于LangChain的带摘要存储对话系统

```python
  from langchain.memory import ConversationSummaryMemory
  from langchain_openai import ChatOpenAI
  from langchain_core.runnables import RunnablePassthrough
  from langchain_core.output_parsers import StrOutputParser
  from langchain_core.prompts import ChatPromptTemplate
  from langchain.chains import LLMChain
  
  # 初始化大模型
  llm = ChatOpenAI(
      model_name = "qwen-plus",
      base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
      api_key="sk-005c3c25f6d042848b29d75f2f020f08",
      temperature=0.7
  )
  
  # 初始化模型和记忆模块
  memory = ConversationSummaryMemory(
      llm=llm,
      memory_key="chat_history",  # 与prompt中的变量名一致
      return_messages=True
  )
  
  # 定义提示模板（必须包含chat_history占位符）
  prompt = ChatPromptTemplate.from_messages([
      ("system", "你是一个助手，需基于对话历史回答问题。当前摘要：{chat_history}"),
      ("human", "{input}")
  ])
  
  # # 创建链并绑定记忆模块
  # chain = LLMChain(
  #     llm=llm,
  #     prompt=prompt,
  #     memory=memory,
  #     verbose=True  # 调试时查看详细流程
  # )
  # 定义LCEL链式流程
  chain = (
      # 注入输入和记忆
      RunnablePassthrough.assign(
          chat_history=lambda _: memory.load_memory_variables({})["chat_history"]
      )
      | prompt  # 将输入传递给提示模板
      | llm  # 调用模型
      | StrOutputParser()  # 解析模型输出为字符串
  )
  
  # 模拟多轮对话
  user_inputs = [
      "人工智能的定义是什么？",
      "小滴课堂上面有什么课程可以学习",
      "人工智能它在医疗领域有哪些应用？"
  ]
  
  for query in user_inputs:
      # 执行对话
      response = chain.invoke({"input": query})
      
      # 打印结果
      print(f"\n用户：{query}")
      print(f"AI：{response}")
      #使用 LLMChain 而非原始LCEL链时，每次 invoke() 会自动调用 memory.save_context()。
      #手动调用场景需显式保存：
      memory.save_context({"input": query}, {"output": response})
      print("当前记忆摘要：", memory.load_memory_variables({})["chat_history"])
  ```

示例输出

```markdown
  用户：人工智能的定义是什么？
  AI：人工智能是模拟人类智能的计算机系统...
  当前记忆摘要： 用户询问人工智能的定义，助手解释了其核心是通过算法模拟人类认知能力。
  
  用户：小滴课堂上面有什么课程可以学习
  AI：如果“小滴课堂”是一个特定的学习平台，建议直接查询该平台提供的课程列表以获取最新信息
  当前记忆摘要： 首先询问了人工智能的定义，AI解释了人工智能是计算机科学的一个分支，随后，人类问到“小滴课堂”上有什么课程可以学习
  
  用户：它在医疗领域有哪些应用？
  AI：在医疗影像分析、药物研发...
  当前记忆摘要： 用户问及医疗应用，助手提到影像分析、药物研发和个性化诊疗是主要方向。
  ```

##### 关键步骤讲解

* **链式组合 (`|` 操作符)**
    通过管道符连接多个组件：
    - `RunnablePassthrough`：传递原始输入
    - `prompt`：格式化提示词
    - `llm`：调用大模型
    - `StrOutputParser`：解析输出为字符串
  * **记忆管理**
    - `memory.load_memory_variables()`：加载当前摘要到提示词
    - `memory.save_context()`：手动保存对话记录（LCEL需要显式保存）
  * **变量绑定**
    使用 `RunnablePassthrough.assign` 动态注入 `chat_history` 变量，确保与提示模板匹配

* 注意： 变量名一致性

- `memory_key` 必须与提示模板中的变量名一致（示例中均为 `chat_history`）。
  - 错误示例：如果模板用 `{summary}` 但 `memory_key` 设为 `history`，会导致变量未注入

##### LCEL与 LLMChain 的核心区别

#### MessagesPlaceholder和多轮AI翻译助手实战

##### MessagesPlaceholder 介绍

* 是 LangChain 中用于在 **聊天型提示模板（ChatPromptTemplate）** 中动态插入消息列表的占位符。

* 允许开发者将历史对话记录、系统消息等结构化地嵌入到 Prompt 中，从而支持多轮对话场景的上下文管理

* 适用场景

* 多角色对话：在聊天机器人中，区分系统指令、用户输入和AI响应。
  * 历史对话注入：将历史消息作为上下文传递给模型，确保对话连贯性。
  * 模块化Prompt设计：灵活组合不同来源的消息（如系统消息、检索结果等）

* 与普通PromptTemplate的区别

* PromptTemplate：用于单字符串模板，适合简单问答。

* ChatPromptTemplate：专为多角色消息设计，必须使用 MessagesPlaceholder 处理消息列表

* 例如

```python
    ChatPromptTemplate.from_messages([
        ("system", "你是一个助手"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}")
    ])
    ```

* 核心功能对比

##### 使用步骤

* 定义模板与占位符

* 在 ChatPromptTemplate 中通过 MessagesPlaceholder 声明占位位置
  * **并指定变量名（需与 Memory 模块的 memory_key 一致）**

```python
  from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
  
  prompt = ChatPromptTemplate.from_messages([
      ("system", "你是一个翻译助手"),
      MessagesPlaceholder(variable_name="chat_history"),
      ("human", "{input}")
  ])
  ```

* 绑定记忆模块

* 使用 ConversationBufferMemory 或 ConversationSummaryMemory 存储对话历史，
  * 并确保 memory_key 与占位符变量名匹配

```python
  from langchain.memory import ConversationBufferMemory
  
  memory = ConversationBufferMemory(
      memory_key="chat_history",  # 必须与MessagesPlaceholder的variable_name一致
      return_messages=True  # 返回消息对象而非字符串
  )
  ```

* 链式调用与历史管理

* 在链式调用中自动注入历史消息，需使用 LLMChain 或 `RunnableWithMessageHistory`

```python
  from langchain.chains import LLMChain
  from langchain_community.chat_models import ChatOpenAI
  
  llm = ChatOpenAI()
  chain = LLMChain(llm=llm, prompt=prompt, memory=memory)
  ```

##### 案例测试

```python
from langchain.memory import ConversationBufferMemory
from langchain_community.chat_models import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chains import LLMChain
from langchain_openai import ChatOpenAI

# 1. 初始化模型与记忆模块
# 初始化大模型
llm = ChatOpenAI(
    model_name = "qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="sk-005c3c25f6d042848b29d75f2f020f08",
    temperature=0.7
)

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 2. 定义包含MessagesPlaceholder的Prompt模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个翻译助手，需参考历史对话优化翻译结果。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "请翻译以下内容：{input}")
])

# 3. 创建链并绑定记忆
chain = LLMChain(llm=llm, prompt=prompt, memory=memory)

# 4. 模拟多轮对话
user_inputs = [
    "Translate 'Hello' to Chinese",
    "Use the translation in a sentence",
    "Now translate 'Goodbye'"
]

for query in user_inputs:
    response = chain.invoke({"input": query})
    print(f"用户：{query}")
    print(f"AI：{response['text']}\n")
    print("当前对话历史：", memory.load_memory_variables({})["chat_history"], "\n")
```

##### LCEL表达式案例

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 3. 创建链并绑定记忆
#chain = LLMChain(llm=llm, prompt=prompt, memory=memory)
#定义LECL表达式，构建chain
chain = (
    RunnablePassthrough.assign(
        chat_history = lambda _ : memory.load_memory_variables({})['chat_history']
    )
    | prompt
    | llm
    | StrOutputParser()    
)

# 4. 模拟多轮对话
user_inputs = [
    "Translate 'Hello' to Chinese",
    "Use the translation in a sentence",
    "Now translate 'Goodbye'"
]

for query in user_inputs:
    response = chain.invoke({"input": query})
    print(f"用户：{query}")
    print(f"AI：{response}\n")
    memory.save_context({"input": query}, {"output": response})
    print("当前对话历史：", memory.load_memory_variables({})["chat_history"], "\n")
```

#### LLM复杂记忆存储-多会话隔离案例实战

##### 背景与需求

* 当多个会话同时与对话系统交互时，需确保每个会话的对话历史独立存储，避免以下问题：

- **数据混淆**：会话A的对话内容泄露给会话B。
  - **上下文丢失**：不同会话的对话历史互相覆盖。
  - **隐私安全**：敏感信息因隔离不当导致泄露。

* **典型场景**

- **客服系统**：不同会话咨询需独立记录。
  - **教育应用**：每个学生与AI助教的对话需单独存档。
  - **医疗助手**：患者健康信息需严格隔离。

* **解决方案：为每个会话分配唯一ID，通过ID隔离的对话历史。**

##### RunnableWithMessageHistory介绍

* 是 LangChain 中用于**动态管理多用户对话历史**的高级封装类，主要解决以下问题

* **会话隔离**：为不同用户/会话（通过 `session_id`）独立存储对话历史。
  * **记忆注入**：自动将历史消息注入到链的每次执行中，无需手动传递。
  * **灵活存储**：支持自定义历史存储后端（内存、数据库、Redis 等）。

* 核心参数

```
  from langchain_core.runnables.history import RunnableWithMessageHistory
  ```

|          参数          |                  类型                   | 必填 |                  说明                   |
  | :--------------------: | :-------------------------------------: | :--: | :-------------------------------------: |
  |       `runnable`       |                Runnable                 |  是  |    基础处理链（需支持消息历史输入）     |
  | `get_session_history`  | Callable[[str], BaseChatMessageHistory] |  是  |  根据session_id获取历史存储实例的函数   |
  |  `input_messages_key`  |                   str                   |  否  |  输入消息在字典中的键名（默认"input"）  |
  | `history_messages_key` |                   str                   |  否  | 历史消息在字典中的键名（默认"history"） |

* 使用场景

* **多用户对话系统**：为每个用户维护独立的对话历史（如客服系统）。
  * **长期会话管理**：结合数据库存储历史，支持跨设备会话恢复。

##### 案例实战

```python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai.chat_models import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 存储会话历史的字典,可以改其他存储结构
store = {}
# 获取会话历史的函数 如果给定的session_id不在store中，则为其创建一个新的ChatMessageHistory实例
def get_session_history(session_id):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 初始化大模型
model = ChatOpenAI(
    model_name = "qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="sk-005c3c25f6d042848b29d75f2f020f08",
    temperature=0.7
)

# 构建聊天提示模板，包含系统消息、历史消息占位符和人类消息
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是一个小滴课堂AI助手，擅长能力{ability}。用30个字以内回答",
        ),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)

# 创建Runnable管道，将提示模板和模型结合在一起
runnable = prompt | model

# 第一次调用带有会话历史的Runnable，提供用户输入和会话ID
response1=with_message_history.invoke(
    {"ability": "Java开发", "input": HumanMessage("什么是jvm")},
    config={"configurable": {"session_id": "user_123"}},#历史信息存入session_id
)
print(f"response1:{response1.content}",end="\n\n")

# 第二次调用带有会话历史的Runnable，用户请求重新回答上一个问题
response2=with_message_history.invoke(
    {"ability": "Java开发", "input": HumanMessage("重新回答一次")},
    config={"configurable": {"session_id": "user_123"}},#历史信息存入session_id，如果改为其他session_id，则不会关联到之前的会话历史
)
print(f"response2:{response2.content}",end="\n\n")

# 打印存储的会话历史
print(f"存储内容:{store}")
```

#### 再进阶复杂存储-多租户多会话隔离案例实战

##### 需求背景

* 系统需要支持多个用户访问，每个用户拥有唯一的标识符（`user_id`）。
* 用户之间的会话历史完全隔离，避免数据混淆。
* 业务场景中，存在一个用户多个会话的场景
* 存储里面需要支持多用户，多会话的方案,  支持不同用户在多个会话中与AI助手交互。

##### 目标

* **多租户支持**：每个用户拥有独立的标识符（`user_id`），确保不同用户之间的数据隔离。

* **多会话支持**：每个用户可以同时维护多个会话（`session_id`），每个会话的历史记录相互独立。

* **灵活扩展性**：支持自定义存储结构和参数配置，便于未来扩展到分布式存储或其他存储介质

* 知识点

* `history_factory_config`

- 是一个配置列表，用于定义额外的可配置字段，（如 `user_id` 和 `session_id`），这些字段可以影响会话历史的生成逻辑。
    - 通过自定义参数来增强系统的灵活性，例如指定用户标识符或对话标识符，从而实现多租户或多会话的支持

* `ConfigurableFieldSpec`

* 是一个类，用于定义一个可配置字段的元信息，包括字段的 ID、类型、名称、描述、默认值等。

```
      from langchain_core.runnables import ConfigurableFieldSpec 
      ```

* 为 `RunnableWithMessageHistory` 提供灵活的参数支持，使得开发者可以通过配置动态调整行为。

| 参数名        | 类型   | 描述                                                         | 示例值                 |
    | :------------ | :----- | :----------------------------------------------------------- | :--------------------- |
    | `id`          | `str`  | 字段的唯一标识符，用于在配置中引用该字段。                   | `"user_id"`            |
    | `annotation`  | `type` | 字段的数据类型，用于类型检查和验证。                         | `str`                  |
    | `name`        | `str`  | 字段的名称，通常用于UI展示或调试信息。                       | `"用户ID"`             |
    | `description` | `str`  | 字段的描述信息，用于说明该字段的用途。                       | `"用户的唯一标识符。"` |
    | `default`     | `any`  | 字段的默认值，如果未提供值时使用。                           | `""`                   |
    | `is_shared`   | `bool` | 是否为共享字段。如果为 `True`，则该字段在整个运行过程中保持不变。 | `True`                 |

##### 解决方案

* 使用 `user_id` 和 `session_id` 组合作为键值，存储在全局字典 `store` 中。
  * 在调用 `get_session_history` 函数时，传入 `user_id` 和 `session_id`，确保获取正确的会话历史。

##### 案例实战

# 存储会话历史的字典,可以改其他存储结构
store = {}
# 获取会话历史的函数 如果给定的session_id不在store中，则为其创建一个新的ChatMessageHistory实例
def get_session_history(user_id: str,session_id: str):
    if (user_id, session_id) not in store:  
        store[(user_id, session_id)] = ChatMessageHistory()  
    return store[(user_id, session_id)]

# 创建Runnable管道，将提示模板和模型结合在一起
runnable = prompt | model

# 创建带有会话历史的Runnable ，使用get_session_history函数来管理会话历史
with_message_history = RunnableWithMessageHistory(
    runnable,
    get_session_history,
    input_messages_key="input", #输入消息在字典中的键名（默认"input"）
    history_messages_key="history", #历史消息在字典中的键名（默认"history"）
    # 定义一些自定义的参数
    history_factory_config=[  
        ConfigurableFieldSpec(  
            id="user_id",  
            annotation=str,  
            name="用户ID",  
            description="用户的唯一标识符。",  
            default="",  
            is_shared=True,  
        ),  
        ConfigurableFieldSpec(  
            id="session_id",  
            annotation=str,  
            name="对话 ID",  
            description="对话的唯一标识符。",  
            default="",  
            is_shared=True,  
        ),  
    ]
)

# 第一次调用带有会话历史的Runnable，提供用户输入和会话ID
response1=with_message_history.invoke(
    {"ability": "Java开发", "input": HumanMessage("什么是jvm")},
    config={'configurable' : { "user_id" : "1" , "session_id" : "1" }}  
)
print(f"response1:{response1.content}",end="\n\n")

# 第二次调用带有会话历史的Runnable，用户请求重新回答上一个问题
response2=with_message_history.invoke(
    {"ability": "Java开发", "input": HumanMessage("重新回答一次")},
    #历史信息存入session_id，如果改为其他session_id，则不会关联到之前的会话历史
    config={'configurable' : { "user_id" : "1" , "session_id" : "2" }},
    
)
print(f"response2:{response2.content}",end="\n\n")
# 打印存储的会话历史
print(f"存储内容:{store}")
  ```

#### 大模型长期记忆解决方案和案例实战

##### 长期记忆的核心需求

* 大模型（LLM）本身不具备长期记忆能力，需通过外部系统实现以下目标：
  * **跨会话记忆持久化**：用户多次对话中产生的关键信息需永久存储。
  * **多用户隔离**：不同用户的对话历史严格隔离，避免数据泄露。
  * **高效检索**：快速提取历史信息辅助当前对话生成。
  * **动态更新**：支持记忆的增量添加和过期清理。
* 解决思路：
  * 持久化存储：将会话历史保存到数据库/缓存，支持跨会话读取。
  * 高效检索：通过语义搜索或键值查询快速定位历史信息。
  * 动态更新：根据新交互持续补充用户画像。

##### LangChain提供的存储方案

* 地址(失效忽略即可)：https://python.langchain.com/docs/integrations/memory/

* `RedisChatMessageHistory` 存储介绍

![1-3677655.png](https://yn-blog.oss-cn-chengdu.aliyuncs.com/agent_v1/2025-09-20/ab4b726a-d59c-47e1-9015-092ca2e8f665.png)

* 内置的 Redis 消息历史存储工具，将会话记录以结构化形式保存至 Redis

* LangChain 的 `RedisChatMessageHistory` 等组件默认依赖 JSON 格式存储对话历史，ReJSON 提供天然的兼容性
  * 数据存储结构

##### 什么是Redis-Stack

* 是Redis 官方推出的 集成化发行版，预装了多个高性能模块和工具，专为现代应用设计。

* **开箱即用**：无需手动安装模块，直接支持搜索、JSON、图数据库等高级功能。

* **开发友好**：更适合需要复杂查询、实时分析、AI 应用的场景（如大模型上下文管理、聊天记录检索）。

* **统一体验**：通过 Redis Stack 命令行或客户端库统一访问所有功能。

* 何时选择 Redis Stack？

* 全文检索（如聊天记录搜索）。
  * 直接操作 JSON 数据（如存储大模型的对话上下文）。
  * 时间序列分析（如监控聊天频率）。
  * 图关系查询（如社交网络分析）。
  * 简化开发流程：避免手动配置多个模块的兼容性和依赖。

* LLM整合如果直接使用之前的Redis服务端会报错，新版单独安装 RedisStack 服务端，才支持相关Redis搜索

* 之前部署的Redis可以卸载，或者更改端口，也可以配置密码，使用方式一样，redis-stack 7.X版本都可以

```dockerfile
  docker run -d \
    --name redis-stack \
    -p 6379:6379 \
    -p 8001:8001 \
    redis/redis-stack:latest
  ```

##### 案例实战

* 安装依赖

```
  pip install langchain-redis==0.2.0 redis==5.2.1
  ```

* 案例测试

```python
  from langchain_redis import RedisChatMessageHistory
  import os
  
  REDIS_URL = os.getenv("REDIS_URL", "redis://47.119.128.20:6379")
  #配置了密码 REDIS_URL = "redis://:your_password@your_ip:6379"
  #REDIS_URL = os.getenv("REDIS_URL", "redis://:abc123456@39.108.115.28:6379")
  
  # 初始化 RedisChatMessageHistory
  history = RedisChatMessageHistory(session_id="user_123", redis_url=REDIS_URL)
  
  # 新增消息
  history.add_user_message("Hello, AI assistant!")
  history.add_ai_message("Hello! How can I assist you today?")
  
  # 检索消息
  print("Chat History:")
  for message in history.messages:
      print(f"{type(message).__name__}: {message.content}")
  ```

* 编码实战

```python
  from langchain_core.chat_history import BaseChatMessageHistory
  from langchain_core.messages import AIMessage, HumanMessage
  from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
  from langchain_core.runnables.history import RunnableWithMessageHistory
  from langchain_openai import ChatOpenAI
  from langchain_redis import RedisChatMessageHistory
  
  #存储历史会话的字典
  
  REDIS_URL="redis://47.119.128.20:6379"
  #定义函数，用于获取会话历史
  def get_redis_history(session_id: str):
      return RedisChatMessageHistory(session_id, redis_url=REDIS_URL)
  
  # 初始化大模型
  model = ChatOpenAI(
      model_name = "qwen-plus",
      base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
      api_key="sk-005c3c25f6d042848b29d75f2f020f08",
      temperature=0.7
  )
  
  # 构建聊天提示模板，包含系统消息、历史消息占位符和人类消息
  prompt = ChatPromptTemplate.from_messages(
      [
          ("system","你是一个小滴课堂AI助手，擅长能力{ability}。用30个字以内回答",),
          MessagesPlaceholder(variable_name="history"),
          ("human", "{input}"),
      ]
  )
  
  #创建基础链
  chain = prompt | model
  
  with_message_history = RunnableWithMessageHistory(
      chain, get_redis_history, input_messages_key="input", history_messages_key="history"
  )
  
  #第一次调用，提供用户输入和会话ID
  resp1 = with_message_history.invoke(
      {"ability":"Java开发", "input":HumanMessage("什么是JVM")}, #替换提示词
      config={"configurable":{"session_id":"user_123"} }) #会话唯一ID
  
  print(f"resp1={resp1.content}",end="\n\n")
  
  resp2 = with_message_history.invoke(
      {"ability":"Java开发", "input":HumanMessage("重新回答一次")}, #替换提示词
      config={"configurable":{"session_id":"user_123"} }) #会话唯一ID
  
  print(f"resp2={resp2.content}",end="\n\n")
  ```

* 注意: 安全起见，RedisStack还是需要配置密码，避免入侵挖矿

```dockerfile
  docker run -d \
    --name redis-stack \
    -p 6379:6379 \
    -p 8001:8001 \
    -e REDIS_ARGS="--requirepass abc123456" \
    redis/redis-stack:latest
  ```

##### 多租户多会话方案

#存储历史会话的字典

REDIS_URL="redis://:abc123456@47.119.128.20:6379"

# 获取会话历史的函数 如果给定的session_id不在store中，则为其创建一个新的ChatMessageHistory实例
def get_session_history(user_id: str,session_id: str):
    uni_key = user_id+"_"+session_id

return RedisChatMessageHistory(
        session_id=uni_key,
        redis_url=REDIS_URL)

# 创建Runnable管道，将提示模板和模型结合在一起
runnable = prompt | model

# 第一次调用带有会话历史的Runnable，提供用户输入和会话ID
response1=with_message_history.invoke(
    {"ability": "Java开发", "input": HumanMessage("什么是jvm")},
    config={'configurable' : { "user_id" : "user_1" , "session_id" : "session_1" }}  
)
print(f"response1:{response1.content}",end="\n\n")
# 打印存储的会话历史
print(get_session_history("user_1","session_1").messages)

# 第二次调用带有会话历史的Runnable，用户请求重新回答上一个问题
response2=with_message_history.invoke(
    {"ability": "Java开发", "input": HumanMessage("重新回答一次")},
    config={'configurable' : { "user_id" : "user_2" , "session_id" : "session_1" }},
)
print(f"response2:{response2.content}",end="\n\n")
# 打印存储的会话历史
print(get_session_history("user_2","session_1").messages)
  ```

评论区