LangChain 记忆与文档处理：让 AI 拥有长期记忆

系列文章：RAG 与 LangChain 开发实战（4/7）
阅读时间：约 25 分钟
前置知识：LangChain 进阶：Chain 链与输出解析器

---

一、会话记忆（Memory）

1.1 为什么需要记忆？

默认情况下，每次调用模型都是独立的：

# 第一次
model.invoke("小明有 2 个猫")  # AI: 好的

# 第二次
model.invoke("总共有几个宠物")  # AI: 我不知道你在说什么

问题：模型不记得之前的对话内容。

解决：使用 LangChain 的 Memory 功能自动附加历史消息。

---

1.2 临时会话记忆（内存存储）

使用 InMemoryChatMessageHistory 在内存中存储历史记录：

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

model = ChatTongyi(model="qwen3-max")

# 提示词模板 - 包含历史占位符
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "根据会话历史回应用户问题。对话历史："),
        MessagesPlaceholder("chat_history"),  # 动态插入历史
        ("human", "请回答：{input}")
    ]
)

# 基础链
base_chain = prompt | model | StrOutputParser()

# 历史存储 - 字典：session_id → InMemoryChatMessageHistory
store = {}

def get_history(session_id):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 增强链 - 自动附加历史
conversation_chain = RunnableWithMessageHistory(
    base_chain,              # 原有链
    get_history,             # 获取历史的函数
    input_messages_key="input",          # 用户输入在模板中的占位符
    history_messages_key="chat_history"  # 历史在模板中的占位符
)

# 使用
session_config = {
    "configurable": {"session_id": "user_001"}
}

conversation_chain.invoke({"input": "小明有 2 个猫"}, session_config)
conversation_chain.invoke({"input": "小刚有 1 只狗"}, session_config)
res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
print(res)  # AI: 总共有 3 个宠物

数据流：

用户输入 → RunnableWithMessageHistory → 自动附加历史 → Prompt → 模型 → 输出
                ↑
            从 store 获取历史

1.3 store 的结构

store = {
    "user_001": InMemoryChatMessageHistory(
        messages=[
            HumanMessage("小明有 2 个猫"),
            AIMessage("好的，记住了"),
            HumanMessage("小刚有 1 只狗"),
            AIMessage("记住了")
        ]
    ),
    "user_002": InMemoryChatMessageHistory(...)  # 另一个会话
}

---

1.4 长期会话记忆（文件存储）

InMemoryChatMessageHistory 的缺点：程序退出后记忆丢失。

解决方案：继承 BaseChatMessageHistory 实现文件存储。

import os
import json
from typing import Sequence
from langchain_core.messages import message_to_dict, messages_from_dict, BaseMessage
from langchain_core.chat_history import BaseChatMessageHistory


class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id, storage_path):
        self.session_id = session_id
        self.storage_path = storage_path
        self.file_path = os.path.join(storage_path, session_id)
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        # 获取已有消息 + 新消息
        all_messages = list(self.messages)
        all_messages.extend(messages)
        
        # 转为字典并保存为 JSON
        new_messages = [message_to_dict(msg) for msg in all_messages]
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump(new_messages, f)

    @property
    def messages(self) -> list[BaseMessage]:
        try:
            with open(self.file_path, "r", encoding="utf-8") as f:
                messages_data = json.load(f)
                return messages_from_dict(messages_data)
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump([], f)


# 使用
def get_history(session_id):
    return FileChatMessageHistory(session_id, "./chat_history")

conversation_chain = RunnableWithMessageHistory(
    base_chain,
    get_history,  # 现在使用文件存储
    input_messages_key="input",
    history_messages_key="chat_history"
)

关键方法：

方法	说明
add_messages()	添加消息到历史
messages	获取历史消息（@property 装饰器）
clear()	清空历史

工具函数：

• message_to_dict()：BaseMessage → dict
• messages_from_dict()：[dict] → [BaseMessage]

---

二、文档加载器（Document Loaders）

2.1 什么是 Document？

Document 是 LangChain 中文档的统一载体：

from langchain_core.documents import Document

doc = Document(
    page_content="文档内容...",  # 文本内容
    metadata={"source": "file.txt", "author": "张三"}  # 元数据
)

2.2 BaseLoader 接口

所有文档加载器都实现 BaseLoader 接口：

from langchain_core.document_loaders import BaseLoader

class BaseLoader:
    def load(self) -> list[Document]:
        """一次性加载全部文档"""
        pass
    
    def lazy_load(self):
        """流式加载，避免内存溢出"""
        pass

2.3 CSVLoader - CSV 文件加载

from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(
    file_path="./data/stu.csv",
    csv_args={
        "delimiter": ",",       # 分隔符
        "quotechar": '"',       # 引用字符
        "fieldnames": ['name', 'age', 'gender']  # 无表头时指定
    },
    encoding="utf-8"
)

# 批量加载
documents = loader.load()  # [Document, Document, ...]

# 懒加载（推荐大文件）
for document in loader.lazy_load():
    print(document)

2.4 JSONLoader - JSON 文件加载

需要额外安装：pip install jq

from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="./data/stu.json",
    jq_schema=".name",      # jq 语法抽取字段
    text_content=False,     # 抽取的是否是字符串
    json_lines=True         # 是否是 JSON Lines 格式
)

documents = loader.load()

jq 语法示例：

• .name - 抽取 name 字段
• .other.addr - 抽取 other 对象下的 addr 字段

2.5 TextLoader - 文本文件加载

from langchain_community.document_loaders import TextLoader

loader = TextLoader("./data/article.txt", encoding="utf-8")
docs = loader.load()  # [Document] - 整个文件一个 Document

---

三、文本分割器（Text Splitters）

3.1 为什么需要分割？

• 大文档超出模型上下文限制
• 向量检索需要更细粒度的文本块
• 保留上下文完整性

3.2 RecursiveCharacterTextSplitter

LangChain 官方推荐的默认分割器：

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每块最大字符数
    chunk_overlap=50,      # 块之间重叠字符数
    separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],
    length_function=len,
)

split_docs = splitter.split_documents(docs)

参数说明：

参数	说明	推荐值
chunk_size	每块最大字符数	500-1000
chunk_overlap	重叠字符数	50-200
separators	分割依据符号	按段落、句子

重叠的作用：保留上下文，语义更完整

块 1: "今天天气真好，阳光明媚，适合出去游玩。"
块 2: "适合出去游玩。下午我们可以去公园散步。"
         ↑ 重叠部分

---

四、向量存储（VectorStores）

4.1 向量存储的作用

存储文本向量，并执行相似性搜索：

┌─────────────────────────────────────────┐
│  向量存储核心 API                         │
├─────────────────────────────────────────┤
│  add_documents()  - 添加文档到向量库     │
│  delete()         - 删除文档             │
│  similarity_search() - 相似度搜索        │
└─────────────────────────────────────────┘

4.2 InMemoryVectorStore - 内存向量存储

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings

vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings(model="text-embedding-v4")
)

# 添加文本
vector_store.add_texts([
    "减肥就是要少吃多练",
    "清淡少油控制卡路里摄入并运动起来",
    "跑步是很好的运动"
])

# 相似度搜索
results = vector_store.similarity_search("怎么减肥？", k=2)
for doc in results:
    print(doc.page_content)

4.3 Chroma - 外部向量数据库

from langchain_chroma import Chroma

vector_store = Chroma(
    collection_name="my_collection",
    embedding_function=DashScopeEmbeddings(model="text-embedding-v4"),
    persist_directory="./chroma_db"  # 持久化路径
)

vector_store.add_documents(documents)
results = vector_store.similarity_search("查询", k=2)

---

五、基于向量检索构建提示词

5.1 基础实现

from langchain_community.chat_models import ChatTongyi
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatTongyi(model="qwen3-max")
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以参考资料为主回答问题。参考资料:{context}"),
        ("user", "用户提问：{input}")
    ]
)

# 初始化向量库
vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings())
vector_store.add_texts(["减肥要少吃多练", "跑步是好运动"])

# 检索
input_text = "怎么减肥？"
results = vector_store.similarity_search(input_text, k=2)

# 构建上下文
context = "[" + "".join([doc.page_content for doc in results]) + "]"

# 执行链
chain = prompt | model | StrOutputParser()
res = chain.invoke({"input": input_text, "context": context})

5.2 使用 RunnablePassthrough 入链

InMemoryVectorStore 不是 Runnable，但可以通过 as_retriever() 转换：

from langchain_core.runnables import RunnablePassthrough
from langchain_core.documents import Document

# 转换为检索器（Runnable）
retriever = vector_store.as_retriever(search_kwargs={"k": 2})

# 文档列表转字符串
def format_docs(docs: list[Document]):
    if not docs:
        return "无相关参考资料"
    return "[" + "".join([doc.page_content for doc in docs]) + "]"

# 构建链
chain = (
    {
        "input": RunnablePassthrough(),      # 透传用户输入
        "context": retriever | format_docs   # 检索 + 格式化
    }
    | prompt
    | model
    | StrOutputParser()
)

res = chain.invoke("怎么减肥？")

数据流解析：

用户输入 ("怎么减肥？")
    ↓
┌───────────────────────────────────┐
│ {"input": RunnablePassthrough(),  │
│  "context": retriever | format}   │
└───────────────────────────────────┘
    ↓
{"input": "怎么减肥？", "context": "[参考资料...]"}
    ↓
prompt → model → 输出

RunnablePassthrough 的作用：

• 透传输入值
• 在字典链中占位，确保输入能同时传给多个分支

---

六、关键知识点总结

6.1 记忆功能

类	存储方式	持久性
InMemoryChatMessageHistory	内存	临时
FileChatMessageHistory	文件	长期

6.2 文档处理

组件	作用
Document	文档统一载体
BaseLoader	文档加载器接口
RecursiveCharacterTextSplitter	文本分割

6.3 向量存储

操作	方法
添加	add_documents() / add_texts()
删除	delete()
检索	similarity_search()
转检索器	as_retriever()

---

七、下一步

现在我们已经掌握了：

• LLM 基础与提示词
• LangChain 模型与模板
• Chain 链与解析器
• 记忆与文档处理
• 向量存储

下一篇我们将深入讲解 RAG 核心概念，理解检索增强生成的完整原理。

---

系列文章导航

1. LLM 基础与提示词工程：从 API 调用到 Prompt 优化技巧
2. LangChain 框架入门：模型抽象与 Prompt 模板
3. LangChain 进阶：Chain 链与输出解析器
4. LangChain 记忆与文档处理：让 AI 拥有长期记忆（本文）
5. RAG 核心概念与向量存储
6. RAG 实战 (上)：知识库构建
7. RAG 实战 (下)：智能客服系统

---

版权声明：本文基于学习笔记整理，转载请注明出处。