LangChain：如何高效管理 LLM 聊天历史记录？

LangChain 团队发布了一篇关于使用 Dragonfly DB 来有效管理 LangChain 应用程序聊天历史记录的教程。

该教程旨在解决用户在使用 LangChain 应用程序时普遍遇到的一个问题：如何高效地管理聊天历史记录。

LangChain 团队在推文中强调了 Dragonfly DB 在管理聊天历史记录中的重要性，并提供了相关教程链接，帮助用户更好地理解和使用 Dragonfly DB。

引言

在快速发展的软件开发领域，为实时 AI 驱动的应用程序（如聊天机器人）优化性能是一项重大挑战。大型语言模型（LLMs）是许多当今高级聊天机器人的核心，但它们本质上是无状态的，因此需要强大的机制来高效管理聊天上下文和会话数据。虽然传统数据库（如 Postgres）能够存储聊天记录，但添加一个具有快速访问速度和多功能数据结构的缓存层对于提高应用程序性能至关重要。

Dragonfly 是一个现代的、多线程的、兼容 Redis 的高性能内存数据存储解决方案，非常适合用于缓存聊天机器人上下文和会话数据。本文探讨了如何集成 Dragonfly 来大幅提升使用 LangChain 构建的聊天机器人的性能，实现对最近聊天会话的快速访问，并确保对话的连续性。所有代码片段可在 dragonfly-examples 仓库中找到。本文将使用 LangChain 库的 Python 版本。

使用 FastAPI 和 LangChain 构建聊天机器人

LangChain 是一个强大的 AI 优先工具包，它简化了使用高级语言模型（如 OpenAI 提供的模型）来创建互动 LLM 应用程序的过程。通过抽象出与这些模型交互的复杂性，LangChain 使开发人员能够专注于优化用户体验和增强对话能力。

考虑一个用户发起对话的简单示例：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

chat = ChatOpenAI(model=MODEL_NAME, temperature=0.2)
chat.invoke(
    [    
        HumanMessage(       
            content="My name is Joe. I would like to learn more about in-memory data stores!"        
        )    
    ]
)

# 内存数据存储是一种数据库管理系统，它将数据存储在计算机的主内存中，而不是在磁盘上。
# 这允许更快地访问数据，因为不需要从磁盘读取或写入数据。

在此场景中，LangChain 处理消息并利用 OpenAI 的 API 生成相关响应。尽管非常强大，但需要注意的是，像 OpenAI 这样的 LLM 本质上是无状态的。在处理初始提示并生成响应后，模型不会保留此交互的任何上下文。如果用户随后提出问题如“我的名字是什么，我想了解什么？”模型将无法回忆起之前的上下文。

这种无状态性在开发需要在多次交互中保持上下文的对话代理时会带来障碍，因为用户期望对话的连续性。如果没有额外层来管理对话上下文（或“状态”，“记忆”），聊天机器人可能会显得脱节，因为它无法识别过去的交互。

存储聊天会话

为了提供无缝和有吸引力的用户体验，必须实现一种机制，使聊天机器人能够记住之前的交互并保持上下文。一种方法是将我们的 LLM 交互包装为后端服务，使用传统数据库如 Postgres 来存储聊天会话和历史记录。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from database import Base

class ChatSession(Base):
    __tablename__ = "chat_sessions"    
    
    id = Column(Integer, primary_key=True)    
    llm_name = Column(String, nullable=False) 
       
    chat_histories = relationship("ChatHistory", back_populates="chat_session")


class ChatHistory(Base):
    __tablename__ = "chat_histories"    
    
    id = Column(Integer, primary_key=True)    
    chat_session_id = Column(Integer, ForeignKey("chat_sessions.id"), nullable=False)    
    is_human_message = Column(Boolean, nullable=False)    
    content = Column(String, nullable=False)   
     
    chat_session = relationship("ChatSession", back_populates="chat_histories")

我们的数据库模式包含两个主要实体：ChatSession 和 ChatHistory。这些实体旨在记录并链接每次聊天会话中的每次交互，使用 SQLAlchemy，一个 Python SQL 工具包和 ORM。如上所示，每个聊天会话都与多个聊天历史记录相关联（多对一），每次需要与 LLM 交互时，我们都会将一堆先前的聊天历史记录（称为 上下文窗口）发送给它。通过这样做，LLM 可以“回忆”此对话的上下文并连续响应。LLM 通常有一个有限的上下文窗口，这意味着我们不能向它们发送无限数量的词或标记。

将一切包装为服务

我们将使用 FastAPI，这是一个现代的、快速的 web 框架，用于使用基于标准 Python 类型提示的 Python 3.6+ 构建 API。我们的第一个 API 端点旨在处理新聊天消息，通过 LangChain 交互 OpenAI API，并使用数据存储维护对话上下文。该端点接收用户发送的 ChatMessageCreate 对象的消息，该对象是用户发送的提示消息。它还有三个依赖项，如你可能猜到的：一个数据库连接，一个 Dragonfly 连接，以及一个由 LangChain 工具包提供的 OpenAI API 客户端。我们将详细介绍数据库和 Dragonfly 交互的更多细节。但这里的主要思想是我们将用户的第一个提示消息发送到 OpenAI，然后创建一个聊天会话记录以及前两个聊天历史记录（即提示和响应）。

@app.post("/chat")
async def new_chat(
       chat_message_human: ChatMessageCreate,       
       db: Session = Depends(get_db_session),       
       df: Dragonfly = Depends(get_dragonfly),       
       chat: ChatOpenAI = Depends(get_chat),
) -> service.ChatSessionResponse:
   # 调用 OpenAI API 获取 AI 响应。   
   message = HumanMessage(content=chat_message_human.content)   
   chat_message_ai = chat.invoke([message])   
   
   # 创建一个新的聊天会话，包含前两个聊天历史记录条目。
   chat_session = ChatSessionCreate(llm_name=LARGE_LANGUAGE_MODEL_NAME)   
   new_chat_histories = __messages_to_histories(chat_message_human, chat_message_ai)   
   srv = service.DataService(db, df)   
   chat_session_response = srv.create_chat_session(chat_session, new_chat_histories)   
   eturn chat_session_response

接下来，我们有以下端点来帮助用户继续与 LLM 的互动。类似的参数和依赖项存在，但这次我们接收聊天会话 ID，它是从上一个端点回复的，以便我们知道用户正在处理特定的聊天会话。我们检查聊天会话的存在性，检索该会话的过去某些数量的聊天历史记录，并将它们与用户新提供的提示消息一起发送给 LLM。通过这样做，LLM 将能够了解对话的上下文并相应地响应。

@app.patch("/chat/{chat_id}")
async def continue_chat(
       chat_id: int,       
       chat_message_human: ChatMessageCreate,       
       db: Session = Depends(get_db_session),       
       df: Dragonfly = Depends(get_dragonfly),       
       chat: ChatOpenAI = Depends(get_chat),
) -> service.ChatSessionResponse:   
   # 检查聊天会话是否存在并加载聊天历史记录。   
   srv = service.DataService(db, df)   
   prev_chat_session_response = srv.read_chat_histories(chat_id)   
   if prev_chat_session_response is None:   
       raise HTTPException(status_code=404, detail="chat not found")   

   # 从聊天历史记录构建消息，然后附加新的用户消息。   
   chat_histories = prev_chat_session_response.chat_histories   
   messages = []   
   for i in range(len(chat_histories)):   
       if chat_histories[i].is_human_message:           
           messages.append(HumanMessage(content=chat_histories[i].content))       
       else:           
           messages.append(AIMessage(content=chat_histories[i].content))   
   messages.append(HumanMessage(content=chat_message_human.content))   
   
   # 调用 OpenAI API 获取 AI 响应。   
   chat_message_ai = chat.invoke(messages)   
   
   # 将两个聊天历史记录条目添加到现有聊天会话中。   
   new_chat_histories = __messages_to_histories(chat_message_human, chat_message_ai)   
   chat_session_response = srv.add_chat_histories(prev_chat_session_response, new_chat_histories)   
   return chat_session_response

一个示例连续上下文对话可能如下所示：

{
    	"chat_session_id": 1,	
    	"chat_histories": [	
    	       { 			
    	               "id": 1,			
    	               "content": "I want to learn more about in-memory data store.",			
    	               "is_human_message": true		
    	       },		
    	       {
                       "id": 2,			
                       "content": "An in-memory data store is a type of database management system that stores data in the main memory 

of a computer rather than on a disk or other storage device...",			
                       "is_human_message": false		},		{			
                       "id": 3,			
                       "content": "What are some good use cases?",			"is_human_message": true		
               },		
               {		
                       "id": 4,			
                       "content": "In-memory data stores are well-suited for a variety of use cases where speed, performance, and real-time data access are critical. Some common use cases for in-memory data stores include:...",			
                       "is_human_message": false		
               }	
       ]
}

最后，我们有一个端点来检索聊天会话。想象一个用户在初次聊天几天后回来，我们仍然能够检索该聊天会话并从那里继续。

缓存最近的聊天会话

缓存是优化服务器端应用程序性能的关键技术。尽管我们无法优化 LLM 的内部工作方式，但有许多机会可以提高服务器的效率。**聊天机器人的交互通常在当前或最近的会话中最为频繁和密集。**用户动态地与机器人互动，通常需要机器人立即回忆对话的上下文。通过缓存这些最近的交互，我们确保聊天机器人能够立即访问会话信息，显著减少检索时间，提升用户体验。

Dragonfly 是缓存的理想解决方案，因为它具有高性能、多线程的能力。它被设计为一个强大的内存数据存储，提供极快的缓存数据访问速度。通过在 Dragonfly 中存储最近聊天会话的上下文和详细信息，我们的聊天机器人可以快速获取必要的信息，而无需反复查询主数据库。如上面的代码片段所示，我们的 DataService 类同时操作数据库和 Dragonfly。以 srv.read_chat_histories(chat_id) 路径为例，我们使用旁路缓存策略，首先尝试从 Dragonfly 读取。如果找到特定 chat_id 的聊天历史记录（作为排序集条目存储），我们可以快速返回响应。如果在 Dragonfly 中找不到键，我们会回退到读取数据库，并被动地在 Dragonfly 中缓存这些历史记录，设置一个过期时间。最近的聊天会话会在 Dragonfly 中随时可用，较旧的会话则不会被丢弃，而是持久存储在数据库中。这些会话由于访问可能性较低而不会主动保存在缓存中。但是，如果一个旧会话再次变得相关（可能是因为用户回到了过去的主题），上述机制会将这些会话从数据库中检索并重新缓存。

def read_chat_histories(self, chat_session_id: int) -> Union[ChatSessionResponse, None]:
    # 检查聊天历史记录是否缓存于 Dragonfly。    
    cache_svc = _DataCacheService(self.df)    
    chat_history_responses = cache_svc.read_chat_histories(chat_session_id)    
    if chat_history_responses is not None and len(chat_history_responses) > 0:    
        return ChatSessionResponse(chat_session_id, chat_history_responses)    
                
    # 如果聊天历史记录未缓存于 Dragonfly，则从数据库读取并缓存于 Dragonfly。    
    chat_histories = self.db.query(models.ChatHistory) \        
        .filter(models.ChatHistory.chat_session_id == chat_session_id) \        
        .order_by(models.ChatHistory.id) \        
        .limit(100) \        
        .all()    
    if chat_histories is None or len(chat_histories) == 0:    
        return None    
        
    chat_history_responses = [ChatHistoryResponse(v.id, v.content, v.is_human_message) for v in chat_histories]    
    cache_svc.add_chat_histories(chat_session_id, chat_history_responses)    
    return ChatSessionResponse(chat_session_id, chat_history_responses)

缓存到磁盘比例：平衡成本和性能

虽然缓存的概念很简单——将数据暂时存储在更快的存储介质中以快速访问，但决定缓存什么数据以及缓存多长时间则更为复杂。一个看似微不足道但常常被忽视的问题是：为什么不缓存所有内容？让我们深入探讨为什么这种方法通常不可行，以及为什么一个出色的驱逐策略很重要。

内存与磁盘的权衡：速度与成本

内存相对于磁盘存储的主要优势在于速度。内存数据的访问时间显著快于磁盘检索。然而，这种速度是有代价的。内存的成本远高于磁盘存储，无论是初始投资还是维护费用。

递减效益

虽然缓存可以显著提升性能，但随着缓存的数据增多，效益会逐渐递减。这是因为并非所有存储的数据都经常被访问。在许多应用中，仅有一小部分数据经常被访问，而大部分数据很少被使用——这被称为 长尾效应，它与我们在聊天机器人示例中看到的情况完全吻合。将这些很少访问的长尾数据存储在昂贵的内存资源中，提供的性能提升相对于成本来说是最小的。

长尾效应

使用 Dragonfly 高效管理缓存

理解这些权衡，Dragonfly 采用了超越传统最近最少使用（LRU）或最少频繁使用（LFU）方法的先进缓存驱逐算法。这些传统算法并不总是与现代应用的实际使用模式相匹配，后者可能需要更不可预测地访问某些类型的数据。

Dragonfly 的驱逐算法旨在通过以下方式智能管理缓存空间：

• 根据访问的最近性和频率优先级数据：确保最相关和最常访问的数据在缓存中停留时间更长。
• 在达到内存限制之前主动驱逐数据：这有助于在不因缓存饱和而突然减速的情况下保持最佳性能。

通过这种方法，Dragonfly 优化了内存使用，确保系统在不增加不必要的内存成本的情况下保持响应性。要利用这一强大功能，只需在启动 Dragonfly 服务器时传递 --cache_mode=true 配置标志。

结论

LangChain 为使用 OpenAI 等 LLM 提供了一个强大的框架，使用 Dragonfly 等工具进行内存管理对于创建互动和连续的用户体验至关重要。通过采用智能缓存策略并维护动态内存数据存储，开发人员可以显著提升聊天机器人的响应速度和上下文感知能力。这不仅改善了用户交互，还优化了资源利用，有效平衡了成本和性能。

需要注意的是，尽管内存解决方案如 Dragonfly 提升了性能，但磁盘数据库对于长期数据持久性和完整性仍然至关重要。它们确保数据随着时间的推移保持安全和可检索，当缓存数据不可用时提供后备支持。缓存策略的探索和聊天会话管理的实际实现表明，通过使用合适的工具和方法，实现无状态 LLM 的有状态交互不仅是可能的，而且是高度有效的。除了先进的缓存驱逐算法，Dragonfly 还有许多吸引人的功能，如完全兼容 Redis 协议、高效的快照机制、Memcached 模式、集群模式等。立即开始使用 Dragonfly（下载社区版或申请免费 Dragonfly Cloud 试用）并构建你自己的惊人 AI 驱动应用程序！

​