当前位置：首页 > news >正文

会话历史管理——持久化

news 来源：原创 2025/5/3 11:35:28

需求场景	推荐方案	理由
中小企业级应用，需复杂查询	MySQL/PostgreSQL	事务支持完善，开发成本低
海量数据高并发写入	Cassandra	水平扩展性强，写入性能高
非结构化历史数据+快速检索	MongoDB	灵活存储，内置全文检索
本地开发或小型应用	SQLite	无需运维，轻量级
语义检索需求	FAISS + 关系型数据库	向量与元数据分离存储，各司其职

ChatMessageHistory

from langchain_community.chat_message_histories import ChatMessageHistory

存储方式：
将对话历史存储在 内存（RAM） 中，数据随程序重启或进程终止而丢失。
特点：
- 简单轻量：无需外部依赖，适合快速原型开发。
- 无持久化：数据不持久保存，仅适用于短期会话。
- 单进程：无法在多个服务实例间共享数据。

Redis

from langchain.memory import RedisChatMessageHistory

存储方式：
将对话历史持久化到 Redis 数据库 中，支持跨会话、跨进程的数据共享。
特点：
- 持久化：数据在服务重启后仍然保留。
- 分布式支持：多个服务实例可访问同一 Redis 集群，适合生产环境。
- 高性能：Redis 作为内存数据库，读写速度极快（微秒级响应）。
- 可扩展性：支持 TTL（自动过期）、备份、集群等高级功能。

LangChain 还支持其他存储后端，可根据需求选择：

DynamoDBChatMessageHistory：AWS DynamoDB 存储。
PostgresChatMessageHistory：PostgreSQL 存储。
FileChatMessageHistory：文件系统存储（JSON 文件）。

MySQL

MySQL 作为会话历史存储

可行性分析

适用性：✅ 完全可行
MySQL 是关系型数据库，适合存储结构化数据（如用户ID、会话ID、消息内容、时间戳等）。
优势：
- 事务支持：保证数据一致性（如消息的原子写入）。
- 复杂查询：支持 SQL 语句灵活查询（如按时间范围、用户ID过滤历史）。
- 持久化：数据可靠存储，支持备份和恢复。
缺点：
- 性能瓶颈：高并发写入时可能需分库分表。
- 扩展性：水平扩展不如 NoSQL 数据库方便。

# Python 代码示例（使用 SQLAlchemy）
from sqlalchemy import create_engine, Column, String, Text, TIMESTAMP
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmakerBase = declarative_base()class ChatHistory(Base):__tablename__ = "chat_history"id = Column(Integer, primary_key=True)session_id = Column(String(36), nullable=False)user_message = Column(Text, nullable=False)bot_message = Column(Text, nullable=False)created_at = Column(TIMESTAMP, default=func.now())# 写入历史记录
def save_history(session_id: str, user_msg: str, bot_msg: str):engine = create_engine("mysql+pymysql://user:password@localhost/db")Session = sessionmaker(bind=engine)with Session() as session:record = ChatHistory(session_id=session_id,user_message=user_msg,bot_message=bot_msg)session.add(record)session.commit()

MySQL+FAISS

若需结合 FAISS 实现语义检索，可采用 混合架构：

元数据存储：使用 MySQL/PostgreSQL 存储会话历史（文本、时间、用户ID）。
向量存储：使用 FAISS 存储消息的向量嵌入，用于语义检索。
关联查询：通过唯一ID关联元数据和向量。

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
import numpy as np# 步骤1：存储元数据到 MySQL
def save_to_mysql(session_id: str, user_msg: str, bot_msg: str):# 参考前文 MySQL 示例代码# 步骤2：生成向量并存储到 FAISS
embeddings = OpenAIEmbeddings()
vector_store = FAISS.load_local("faiss_index")def save_to_faiss(text: str, metadata: dict):embedding = embeddings.embed_query(text)vector_store.add_embeddings([embedding], [metadata])vector_store.save_local("faiss_index")# 关联元数据与向量
metadata = {"session_id": "abc123", "message_id": 1}
save_to_faiss("用户的问题文本", metadata)
save_to_mysql("abc123", "用户的问题文本", "AI的回答文本")