2026-04-18发表2026-04-19更新AI / LangChain26 分钟读完 (大约3836个字)

LangChain：记忆组件(Memory)的实现和使用

本文档基于 LangChain v0.3+ / v1.0 官方推荐，聚焦于 BaseChatMessageHistory 及其主流持久化实现，以LangGraph 记忆、通用集成模式、记忆管理策略等，并更新了决策树和推荐路径。新项目应统一使用 LCEL + RunnableWithMessageHistory 构建有状态对话系统。

已弃用的旧版记忆类（如 ConversationBufferWindowMemory、ConversationSummaryMemory、VectorStoreRetrieverMemory 等）不再推荐使用。

记忆组件

LangChain 最新的记忆组件遵循着一个统一的设计：所有实现都继承自 BaseChatMessageHistory 基类。在开发或测试中，可以直接使用内存版本 InMemoryChatMessageHistory，而生产环境则需要搭配 RunnableWithMessageHistory 这个核心组件，将聊天历史无缝接入你的 LLM 应用链中。

主流的记忆组件：

这些是 LangChain 推荐在生产环境中使用的、积极维护的记忆组件。

组件	核心功能	维护状态	导入方式
BaseChatMessageHistory	抽象基类，所有聊天历史实现都必须继承它	核心组件	`from langchain_core.chat_history import BaseChatMessageHistory`
RunnableWithMessageHistory	核心包装器，负责管理和持久化聊天历史	核心组件	`from langchain_core.runnables.history import RunnableWithMessageHistory`
InMemoryChatMessageHistory	内存存储，主要用于开发和测试	核心组件	`from langchain_core.chat_history import InMemoryChatMessageHistory`
RedisChatMessageHistory	Redis 存储，利用 Redis 的高性能和 JSON 搜索能力	langchain_redis	`from langchain_redis import RedisChatMessageHistory`
PostgresChatMessageHistory	PostgreSQL 存储，通过 `psycopg 3` 驱动支持异步操作	langchain_postgres	`from langchain_postgres import PostgresChatMessageHistory`
FileChatMessageHistory	文件存储，将聊天历史保存为本地 JSON 文件	langchain_community	`from langchain_community.chat_message_histories import FileChatMessageHistory`
CassandraChatMessageHistory	Apache Cassandra 存储	langchain_community	`from langchain_community.chat_message_histories import CassandraChatMessageHistory`
MongoDB ChatMessageHistory	MongoDB 存储	langchain_mongodb	`from langchain_mongodb.chat_message_histories import MongoDBChatMessageHistory`
SQLChatMessageHistory	SQL 数据库存储	langchain_community	`from langchain_community.chat_message_histories import SQLChatMessageHistory`

LangChain 官方社区还提供了大量与外部服务集成的记忆组件，方便连接各种云数据库和存储服务。这些集成组件大多位于 langchain_community 包下，使用时需要注意其具体的维护状态。

云数据库与存储：

DynamoDBChatMessageHistory：用于 AWS DynamoDB
CosmosDBChatMessageHistory：用于 Azure CosmosDB
FirestoreChatMessageHistory：用于 Google Firestore
UpstashRedisChatMessageHistory：用于 Upstash Redis
MomentoChatMessageHistory：用于 Momento 缓存服务
XataChatMessageHistory：用于 Xata 数据库
TiDBChatMessageHistory：用于 TiDB 数据库

消息流与向量数据库：

KafkaChatMessageHistory：用于 Apache Kafka 消息队列
ZepChatMessageHistory：用于 Zep 向量数据库
RocksetChatMessageHistory：用于 Rockset 实时分析数据库

抽象基类

LangChain 的记忆组件都继承自 BaseChatMessageHistory 抽象基类，定义了统一的接口规范。实现类需要覆写以下核心方法：

方法	用途
`add_messages(messages)` / `aadd_messages(messages)`	批量添加消息
`messages` / `aget_messages()`	获取所有消息
`clear()` / `aclear()`	清空会话消息

使用规范：更新历史时应优先使用 add_messages（批量添加），而非 add_message、add_user_message 或 add_ai_message 等逐条添加方式，以避免对底层持久化层产生不必要的往返调用。

内存实现

InMemoryChatMessageHistory 是 BaseChatMessageHistory 的内存实现，直接将消息存储在 Python 列表（list）中。

核心参数

参数/属性	类型	说明
`messages`	`List[BaseMessage]`	存储消息的列表

核心方法

方法	说明
`add_messages(messages)`	批量添加消息到内存列表
`aadd_messages(messages)`	异步批量添加消息
`clear()` / `aclear()`	清空所有消息
`add_user_message()` / `add_ai_message()`	便捷方法，推荐优先使用 `add_messages`

使用示例

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

history = InMemoryChatMessageHistory()
history.add_messages([
    HumanMessage(content="Hello!"),
    AIMessage(content="Hi! How can I help?")
])
messages = history.messages
history.clear()

适用场景

本地开发和测试：无需外部依赖，快速验证原型
单用户、临时对话：应用重启后历史即消失

Redis 实现

RedisChatMessageHistory 是基于 Redis 的聊天消息历史存储实现，利用 Redis JSON 存储消息，并通过 RedisVL 提供高效的索引和查询功能。

核心参数

参数	类型	默认值	说明
`session_id`	`str`	必填	会话唯一标识符
`redis_url`	`str`	`'redis://localhost:6379'`	Redis 实例连接 URL
`key_prefix`	`str`	`'chat:'`	Redis 键名前缀，用于隔离不同应用的命名空间
`ttl`	`int	None`	`None`
`index_name`	`str`	`'idx:chat_history'`	Redis 搜索索引名称
`redis_client`	`Redis	None`	`None`
`overwrite_index`	`bool`	`False`	是否覆盖已存在的索引

核心方法

方法	说明
`add_user_message(message)`	添加用户消息
`add_ai_message(message)`	添加 AI 消息
`add_message(message)`	添加通用消息
`messages`（属性）	获取该会话的所有消息
`search_messages(query, limit=10)`	对消息内容进行全文搜索，支持 RedisVL 的 TextQuery 能力
`clear()`	清空当前会话的所有消息

数据存储结构

每条消息在 Redis 中以 JSON 格式存储，键名模式为 {key_prefix}{session_id}:{timestamp}，存储字段如下：

字段	说明
`type`	消息类型（human / ai / system）
`data.content`	实际消息文本
`data.additional_kwargs`	附加参数
`session_id`	会话标识符
`timestamp`	Unix 时间戳

使用示例

from langchain_redis import RedisChatMessageHistory

history = RedisChatMessageHistory(
    session_id="user_123",
    redis_url="redis://localhost:6379",
    key_prefix="myapp:chat:",
    ttl=3600,
    index_name="my_custom_idx"
)

history.add_user_message("Hello, AI assistant!")
history.add_ai_message("Hello! How can I assist you today?")
all_messages = history.messages
results = history.search_messages("artificial intelligence", limit=5)
history.clear()

核心特性

特性	说明
会话隔离	通过 `session_id` 实现多会话独立存储
自动索引	自动创建 Redis 搜索索引，支持高效消息检索
全文搜索	`search_messages()` 方法提供消息内容全文搜索能力
TTL 支持	消息自动过期，避免历史无限增长
低延迟	Redis 内存特性保证毫秒级读写，适合高并发场景

适用场景

生产环境高并发应用：会话数量大、读写频繁的聊天机器人
需要历史搜索功能：对聊天记录进行全文检索的场景
分布式部署：多个应用实例共享同一 Redis 集群
需要消息自动过期：控制存储成本

PostgreSQL 实现

PostgresChatMessageHistory 是 langchain-postgres 包提供的 PostgreSQL 聊天消息历史持久化实现，支持同步和异步操作，底层使用 psycopg 3 驱动。

核心参数

参数	类型	说明
`table_name`	`str`	存储聊天历史的表名
`session_id`	`str`	会话唯一标识符
`sync_connection`	`psycopg.Connection	None`
`async_connection`	`psycopg.AsyncConnection	None`

核心方法

方法	异步版本	说明
`create_tables(connection, table_name)`	`acreate_tables()`	静态方法，创建表结构
`drop_table(connection, table_name)`	`adrop_table()`	静态方法，删除表
`add_messages(messages)`	`aadd_messages()`	添加消息到历史
`get_messages()`	`aget_messages()`	获取会话的所有消息
`clear()`	`aclear()`	清空会话消息
`messages`（属性）	-	返回会话的所有消息

数据库 Schema

字段	类型	说明
`session_id`	UUID 或文本	会话标识符，自动建立索引优化查询性能
`message`	JSONB	序列化的消息内容（类型、内容、附加属性）

消息序列化使用 message_to_dict 函数转换为 JSON，读取时通过 messages_from_dict 反序列化。

同步使用示例

from langchain_postgres import PostgresChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage
import psycopg

conn = psycopg.connect("postgresql://user:pass@localhost/db")
PostgresChatMessageHistory.create_tables(conn, "chat_history")

history = PostgresChatMessageHistory(
    table_name="chat_history",
    session_id="user_123",
    sync_connection=conn
)

history.add_messages([
    HumanMessage(content="Hello!"),
    AIMessage(content="Hi there!")
])
messages = history.messages
history.clear()

异步使用示例

import asyncio
import psycopg
from langchain_postgres import PostgresChatMessageHistory

async def main():
    conn = await psycopg.AsyncConnection.connect("postgresql://user:pass@localhost/db")
    await PostgresChatMessageHistory.acreate_tables(conn, "chat_history")
    history = PostgresChatMessageHistory(
        table_name="chat_history",
        session_id="user_123",
        async_connection=conn
    )
    await history.aadd_messages([HumanMessage(content="Hello async!")])
    messages = await history.aget_messages()

asyncio.run(main())

适用场景

需要强一致性的生产应用：ACID 事务保证数据完整性
已有 PostgreSQL 基础设施：复用现有数据库，降低运维复杂度
需要复杂查询和分析：利用 SQL 能力分析对话数据
需要消息永久存储：无自动过期，数据长期保存

三种实现对比

对比维度	InMemoryChatMessageHistory	RedisChatMessageHistory	PostgresChatMessageHistory
存储介质	内存（Python list）	Redis（内存 + 持久化）	PostgreSQL（磁盘持久化）
持久化	❌ 应用重启丢失	✅ 支持	✅ 支持
并发性能	单进程内，无跨进程共享	极高（内存级 I/O）	中等
全文搜索	❌	✅ `search_messages()`	❌（需自行实现）
TTL/过期	❌	✅ `ttl` 参数	❌（需手动清理）
跨进程共享	❌	✅	✅
异步支持	✅	✅	✅（psycopg 3）
部署复杂度	最低（无依赖）	中等（需 Redis 服务）	中等（需 PostgreSQL）
适用场景	本地开发、单用户、临时测试	生产高并发、分布式、需要历史搜索	生产环境、强一致性、已有 PostgreSQL

选型建议

开发/测试阶段：InMemoryChatMessageHistory 零依赖，快速验证
生产高并发服务：RedisChatMessageHistory 性能最优，适合分布式部署
强一致性/审计需求：PostgresChatMessageHistory 提供 ACID 事务保证
需要历史搜索：RedisChatMessageHistory 内置全文搜索能力

通用集成模式

所有实现均遵循 BaseChatMessageHistory 接口，因此可与 RunnableWithMessageHistory 无缝集成。

以下示例使用内存存储（InMemoryChatMessageHistory）作为后端，展示如何构建一个带对话记忆的聊天链。

# 导入所需的类和函数
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI  # 使用 OpenAI 的聊天模型

# 1. 定义一个存储会话历史的字典（生产环境可使用 Redis/PostgreSQL 等持久化存储）
#    键为 session_id，值为 InMemoryChatMessageHistory 实例
store = {}

# 2. 定义一个函数，根据 session_id 获取对应的聊天历史对象
def get_session_history(session_id: str):
    """返回给定 session_id 的 BaseChatMessageHistory 实例"""
    if session_id not in store:
        # 如果该会话不存在，则创建新的内存历史记录对象
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 3. 构建一个基本的 LCEL 链（LangChain Expression Language）
#    创建一个提示模板，其中包含一个名为 "history" 的 MessagesPlaceholder，
#    用于动态注入历史消息，以及一个 "input" 变量用于接收用户当前输入。
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Answer the user's questions concisely."),
    MessagesPlaceholder(variable_name="history"),  # 历史消息将插入此处
    ("human", "{input}")                           # 用户当前输入
])

# 4. 初始化聊天模型（这里使用 OpenAI GPT-3.5-turbo，请确保已设置 OPENAI_API_KEY）
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# 5. 构建输出解析器，将模型输出解析为字符串
parser = StrOutputParser()

# 6. 将提示模板、模型和解析器组合成一个完整的链
chain = prompt | model | parser

# 7. 使用 RunnableWithMessageHistory 包装上述链，使其具备自动管理历史消息的能力
#    参数说明：
#    - runnable: 基础链
#    - get_session_history: 用于获取会话历史的函数
#    - input_messages_key: 输入字典中代表用户当前输入的键名（与提示模板中的变量名一致）
#    - history_messages_key: 输入字典中用于传递历史消息的键名（与 MessagesPlaceholder 的 variable_name 一致）
chain_with_history = RunnableWithMessageHistory(
    runnable=chain,
    get_session_history=get_session_history,
    input_messages_key="input",      # 对应提示模板中的 "input"
    history_messages_key="history",  # 对应 MessagesPlaceholder 的 "history"
)

# 8. 调用带记忆的链
#    第一次调用：没有历史消息，模型将基于系统提示和当前输入回答
config = {"configurable": {"session_id": "user_123"}}  # 配置中指定 session_id
response1 = chain_with_history.invoke(
    {"input": "Hi, my name is Alice."},
    config=config
)
print("AI 第一次回复:", response1)

# 9. 第二次调用：使用相同的 session_id，历史消息会自动被加载并注入到提示中
response2 = chain_with_history.invoke(
    {"input": "What is my name?"},
    config=config
)
print("AI 第二次回复:", response2)

# 10. 可选：查看当前会话存储的历史消息（验证记忆是否生效）
print("\n--- 当前会话的历史消息 ---")
for msg in store["user_123"].messages:
    print(f"{msg.type}: {msg.content}")

# 11. 清空某个会话的历史（如果需要）
# store["user_123"].clear()

运行结果

AI 第一次回复: Hello Alice! How can I help you today?
AI 第二次回复: Your name is Alice.
--- 当前会话的历史消息 ---
human: Hi, my name is Alice.
ai: Hello Alice! How can I help you today?
human: What is my name?
ai: Your name is Alice.

代码说明

**get_session_history**：必须返回一个 BaseChatMessageHistory 子类的实例。示例代码使用内存存储，生产环境可替换为 RedisChatMessageHistory 或 PostgresChatMessageHistory。
**MessagesPlaceholder**：在提示模板中定义一个占位符（如 "history"），RunnableWithMessageHistory 会自动将之前的历史消息填充到该位置。
**config**：调用时通过 configurable 字段传递 session_id，框架会自动根据该 ID 获取对应的历史记录。
**input_messages_key 和 history_messages_key**：必须与提示模板中的变量名精确匹配，否则无法正确注入。

LangGraph中的记忆

LangGraph 是 LangChain 的 Agent 编排框架，其记忆架构分为两层：

短期记忆（Short-Term Memory）：通过 checkpointer 机制实现，作用于单个会话（thread），用于保存对话连续性。
长期记忆（Long-Term Memory）：通过 BaseStore 接口实现，支持跨会话的知识共享和持久化存储。

使用 MongoDB Store 的示例：

from langgraph.store.mongodb import MongoDBStore

store = MongoDBStore(
    connection_string="mongodb://localhost:27017",
    collection_name="agent_memories",
    ttl_seconds=86400   # 24小时后自动过期
)

TTL（Time-To-Live） 是 LangGraph 记忆管理的核心配置，合理设置可自动清理过期数据。

记忆管理策略

策略	说明
写入（Write）	创建清晰、结构化的记忆内容
选择（Select）	仅保留与当前任务最相关的信息
压缩（Compress）	通过摘要或截断减少 token 占用
隔离（Isolate）	将不同类型记忆分离管理

混合记忆架构：组合使用短期记忆（如 InMemoryChatMessageHistory 配合 trim_messages）和长期记忆（如 RedisChatMessageHistory 配合语义检索），在多轮对话中兼顾即时上下文的连续性和历史知识的检索能力。

第三方集成

除内置存储实现外，LangChain 生态支持多种第三方集成：

集成名称	说明
`langchain-recallio`	基于 Recallio 云服务的记忆，支持向量语义检索、TTL 和标签过滤
`langgraph-checkpoint-aws`	使用 AWS Bedrock Session Management Service 实现的检查点
`MongoDB 集成`	`langgraph-store-mongodb`，支持跨会话长期记忆和 Atlas Vector Search
`向量数据库`	Pinecone、Weaviate、Chroma、FAISS、Milvus 等可作为语义检索后端，与 `RunnableWithMessageHistory` 组合使用

官方推荐路径

阶段	推荐方案
开发/测试	`InMemoryChatMessageHistory` + `LCEL` + `RunnableWithMessageHistory`
单会话生产（高性能）	`RedisChatMessageHistory` + `RunnableWithMessageHistory`
单会话生产（强一致）	`PostgresChatMessageHistory` + `RunnableWithMessageHistory`
需要语义检索长期记忆	向量数据库 + `VectorStoreRetriever` + `RunnableWithMessageHistory`
Agent 系统	`LangGraph` + `BaseStore`（长期）+ `Checkpointer`（短期）

完整的记忆系统应综合运用存储、检索、压缩、清理多种策略，并始终以 LCEL（RunnableWithMessageHistory）为统一的集成入口。

LangChain：记忆组件(Memory)的实现和使用

http://blog.gxitsky.com/2026/04/18/AI-LangChain-039-Chain-Memory/

作者

光星

发布于

2026-04-18

更新于

2026-04-19

LangChain：记忆组件(Memory)的实现和使用

记忆组件

抽象基类

内存实现

核心参数

核心方法

使用示例

适用场景

Redis 实现

核心参数

核心方法

数据存储结构

使用示例

核心特性

适用场景

PostgreSQL 实现

核心参数

核心方法

数据库 Schema

同步使用示例

异步使用示例

适用场景

三种实现对比

选型建议

通用集成模式

LangGraph中的记忆

记忆管理策略

第三方集成

官方推荐路径

作者

发布于

更新于

许可协议

喜欢这篇文章？打赏一下作者吧

评论

目录