LangChain 聊天历史

孟宝亮2025-04-13机器学习/课程内容67

在使用 LangChain 构建对话系统时，聊天记录的管理、处理是至关重要的一环。本篇文章主要介绍如何存储对话历史，以及如何对对话历史进行处理。

1. 对话历史存储

LangChain 提供了多种灵活的聊天历史记录管理方式，以满足不同场景下的需求。从最简单的内存存储（ChatMessageHistory），到将记录持久化到本地文件（FileChatMessageHistory），再到支持多用户和高并发场景的 Redis 存储（RedisChatMessageHistory），以及适用于结构化存储和查询的 SQL 数据库存储（SQLChatMessageHistory），这些类都封装了便捷的方法来添加消息、持久化数据，并支持随时访问历史对话内容。本文将依次介绍这几种历史记录管理方式的使用方法及其适用场景，帮助开发者根据实际需求灵活选择合适的方案。

1.1 ChatMessageHistory

使用 langchain_community.chat_message_histories.ChatMessageHistory 类来管理和操作聊天记录。具体作用如下：

  1. 创建历史记录对象：通过 ChatMessageHistory() 创建了一个聊天历史记录对象 history。
  2. 添加用户和AI的消息：
    • history.add_user_message('中国的首都是哪里？') 添加了用户的一条消息。
    • history.add_ai_message('北京。') 添加了AI的回应。
    • history.add_user_message('美国的首都是哪里？') 添加了用户的第二条消息。
  3. 打印所有历史对话：通过 print(history.messages) 输出了聊天记录，包括用户和AI的所有消息。

from langchain_community.chat_message_histories import ChatMessageHistory


def demo():
    history = ChatMessageHistory()
    history.add_user_message('中国的首都是哪里？')
    history.add_ai_message('北京。')
    history.add_user_message('美国的首都是哪里？')

    # 获得所有的历史对话
    print(history.messages)


if __name__ == '__main__':
    demo()

1.2 FileChatMessageHistory

使用 langchain_community.chat_message_histories.FileChatMessageHistory 类来管理和保存聊天记录到文件。具体作用如下：

  1. 创建文件历史记录对象：通过 FileChatMessageHistory(file_path='chat_history.txt') 创建了一个聊天历史记录对象 history，并指定了一个文件路径 'chat_history.txt'，用于保存聊天记录。
  2. 添加用户和AI的消息：
    • history.add_user_message('中国的首都是哪里？') 添加了用户的一条消息。
    • history.add_ai_message('北京。') 添加了AI的回应。
    • history.add_user_message('美国的首都是哪里？') 添加了用户的第二条消息。
  3. 打印所有历史对话：通过 print(history.messages) 输出了聊天记录，包括用户和AI的所有消息。

这段代码的目的是展示如何通过 FileChatMessageHistory 将聊天记录保存在指定的文件中，并且能够在需要时读取和查看这些记录。

from langchain_community.chat_message_histories import FileChatMessageHistory


def demo():
    history = FileChatMessageHistory(file_path='chat_history.txt')
    history.add_user_message('中国的首都是哪里？')
    history.add_ai_message('北京。')
    history.add_user_message('美国的首都是哪里？')

    print(history.messages)


if __name__ == '__main__':
    demo()

1.3 RedisChatMessageHistory

这段代码展示了如何使用 langchain_community.chat_message_histories.RedisChatMessageHistory 类通过 Redis 存储聊天记录。具体作用如下：

  1. 安装和配置 Redis：
    • 提供了安装 Redis 服务的命令，如 pip install redis 和 Redis 服务的启动命令。
    • 配置 Redis 以允许通过 0.0.0.0 绑定访问（使其可以从其他设备连接）。
  2. 创建 Redis 聊天历史记录对象：
    • 通过 RedisChatMessageHistory(session_id='user_001', url='redis://127.0.0.1:6379/0', key_prefix='chat_history:') 创建了一个 Redis 聊天历史记录对象 history，并指定了 Redis 连接 URL 和一个会话 ID（user_001），用作唯一标识符。
    • key_prefix='chat_history:' 用来为存储的键添加前缀。
  3. 添加用户和AI的消息：
    • history.add_user_message('中国的首都是哪里？') 添加了用户的一条消息。
    • history.add_ai_message('北京。') 添加了 AI 的回应。
    • history.add_user_message('美国的首都是哪里？') 添加了用户的第二条消息。
  4. 打印所有历史对话：
    • 通过 print(history.messages) 输出了从 Redis 中获取的聊天记录，包括用户和 AI 的所有消息。

这段代码的目的是展示如何通过 Redis 存储和管理聊天历史记录，并且可以在需要时查看这些记录。

from langchain_community.chat_message_histories import RedisChatMessageHistory
# pip install redis
# sudo apt install redis-server
# sudo service redis-server start
# sudo service redis-server status
# sudo vim /etc/redis/redis.conf
# bind 127.0.0.1 ::1  修改为 bind 0.0.0.0
# sudo service redis-server restart
# FLUSHALL 清空数据


def demo():
    history = RedisChatMessageHistory(session_id='user_001', url='redis://127.0.0.1:6379/0', key_prefix='chat_history:')
    history.add_user_message('中国的首都是哪里？')
    history.add_ai_message('北京。')
    history.add_user_message('美国的首都是哪里？')
    print(history.messages)


if __name__ == '__main__':
    demo()

1.4 SQLChatMessageHistory

这段代码展示了如何使用 langchain_community.chat_message_histories.SQLChatMessageHistory 类通过 SQLite 数据库存储和管理聊天记录。具体作用如下：

  1. 创建 SQLite 数据库连接：
    • database = 'sqlite:///memory.db' 定义了一个 SQLite 内存数据库连接，这意味着数据库将存储在内存中，而不是一个物理文件。memory.db 是数据库的名称。
  2. 创建 SQL 聊天历史记录对象：
    • 通过 SQLChatMessageHistory(session_id='user_002', connection=database, table_name='chat_history') 创建了一个聊天历史记录对象 history，并指定了：
      • session_id='user_002'：表示此会话的唯一标识符。
      • connection=database：连接到上面创建的 SQLite 数据库。
      • table_name='chat_history'：指定表格的名称，用于存储聊天记录。
  3. 添加用户和AI的消息：
    • history.add_user_message('中国的首都是哪里？') 添加了用户的一条消息。
    • history.add_ai_message('北京。') 添加了 AI 的回应。
    • history.add_user_message('美国的首都是哪里？') 添加了用户的第二条消息。
  4. 打印所有历史对话：
    • 通过 print(history.messages) 输出了从 SQLite 数据库中获取的聊天记录，包括用户和 AI 的所有消息。

这段代码的目的是展示如何通过 SQLite 数据库来存储和管理聊天历史记录，并能在需要时查询这些记录。

from langchain_community.chat_message_histories import SQLChatMessageHistory


def demo():
    database = 'sqlite:///memory.db'
    history = SQLChatMessageHistory(session_id='user_002', connection=database, table_name='chat_history')
    history.add_user_message('中国的首都是哪里？')
    history.add_ai_message('北京。')
    history.add_user_message('美国的首都是哪里？')
    print(history.messages)


if __name__ == '__main__':
    demo()

2. 对话历史处理

在对话系统中，高效地管理和处理消息内容是实现流畅交互的关键。LangChain 提供了丰富的工具函数，如 trim_messages 和 filter_messages，用于裁剪和筛选聊天记录，帮助开发者更灵活地控制上下文的长度和内容。无论是为了满足大模型的 token 限制，还是为了提取特定角色或消息类型的内容，这些工具都提供了简洁而强大的支持。本文将通过具体示例介绍如何使用这些函数，实现基于 token 数量或消息属性的裁剪与过滤，助你更好地构建智能对话逻辑。

2.1 trim_messages

这段代码展示了如何使用 langchain_core.messages.trim_messages 函数根据不同策略裁剪聊天记录。

from langchain_core.messages import trim_messages
from langchain.schema import AIMessage, HumanMessage, SystemMessage
from langchain_core.messages.utils import count_tokens_approximately
from transformers import AutoTokenizer


messages = [
    SystemMessage("你是一个幽默的助手，总是用笑话回答。", id='0', name='system'),
    HumanMessage("为什么它叫 langchain？", id='1', name='trump'),
    AIMessage("可能他们觉得 'WordRope' 或 'SentenceString' 太没感觉了！", id='2', name='obama'),
    HumanMessage("那 Harrison 追的是谁？", id='3', name='trump'),
    AIMessage("我想想，大概是在追办公室里最后一杯咖啡！", id='4', name='obama'),
    HumanMessage("你怎么称呼一只不会说话的鹦鹉？", id='5', name='trump'),
]


# 1. 根据标记裁剪
def demo01():

    # 自定义 token 计算
    tokenizer = AutoTokenizer.from_pretrained('deepseek')
    def custom_calculate_length(message):
        tokens = tokenizer.tokenize(message[0].content)
        return len(tokens)

    trim = trim_messages(messages,
                         strategy='last',
                         token_counter=custom_calculate_length,
                         max_tokens=50,
                         start_on=('human', 'ai'),
                         end_on=('human', 'ai'),
                         include_system=True)
    print(trim)


# 2. 根据消息裁剪
def demo02():
    # 每一条完整的消息作为一个 token
    # token_counter=len,
    trim = trim_messages(messages,
                         strategy='last',
                         token_counter=len,
                         max_tokens=3,
                         start_on=('human', 'ai'),
                         end_on=('human', 'ai'),
                         include_system=True)
    print(trim)


if __name__ == '__main__':
    demo01()
    demo02()

2.2 filter_messages

这段代码展示了如何使用 langchain_core.messages.filter_messages 函数对聊天记录进行过滤。

from langchain_core.messages import filter_messages
from langchain.schema import AIMessage, HumanMessage, SystemMessage
from langchain_core.messages.utils import count_tokens_approximately
from transformers import AutoTokenizer


messages = [
    SystemMessage("你是一个幽默的助手，总是用笑话回答。", id='0', name='system'),
    HumanMessage("为什么它叫 langchain？", id='1', name='trump'),
    AIMessage("可能他们觉得 'WordRope' 或 'SentenceString' 太没感觉了！", id='2', name='obama'),
    HumanMessage("那 Harrison 追的是谁？", id='3', name='trump'),
    AIMessage("我想想，大概是在追办公室里最后一杯咖啡！", id='4', name='obama'),
    HumanMessage("你怎么称呼一只不会说话的鹦鹉？", id='5', name='trump'),
]


def demo():
    filtered_messages = filter_messages(messages, include_types='human')
    print(filtered_messages)

    print('-' * 100)
    filtered_messages = filter_messages(messages, exclude_types='human')
    print(filtered_messages)

    print('-' * 100)
    filtered_messages = filter_messages(messages, include_names='obama')
    print(filtered_messages)

    print('-' * 100)
    filtered_messages = filter_messages(messages, exclude_ids=['1', '2'])
    print(filtered_messages)


if __name__ == '__main__':
    demo()

3. 多轮对话案例

from dotenv import load_dotenv
load_dotenv('llm.env')
from langchain_deepseek import ChatDeepSeek
from langchain.prompts import ChatPromptTemplate
from langchain.prompts import MessagesPlaceholder
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_core.messages import trim_messages


def demo():

    model = ChatDeepSeek(model='deepseek-chat')
    history = ChatMessageHistory()
    parser = StrOutputParser()


    prompt = ChatPromptTemplate.from_messages([
        ('system', 'You are a helpful assistant.'),
        MessagesPlaceholder(variable_name='chat_history'),
        ('user', '{user_input}'),
    ])

    def show_history(messages):
        for message in messages:
            print(message.content)
        print('-' * 100)

    def get_history():
        messages = trim_messages(history.messages,
                                 max_tokens=4,
                                 token_counter=len,
                                 strategy='last',
                                 start_on=('human', 'ai'),
                                 end_on=('human', 'ai'))
        show_history(messages)
        return messages

    chain = {'user_input': RunnablePassthrough(), 'chat_history': lambda _: get_history()} | prompt | model | parser

    while True:
        # 获得用户输入
        user_input = input('用户输入:')
        # 生成聊天回复
        response = chain.invoke(user_input)
        print('助手回复:', response)
        # 记录聊天历史
        history.add_user_message(user_input)
        history.add_ai_message(response)

        # show_history(history.messages)


if __name__ == '__main__':
    demo()
未经允许不得转载：一亩三分地 » LangChain 聊天历史
0 赞
评论 (0)

5 + 9 =

文章目录

《LangChain 开发实战》

©2024 一亩三分地 京ICP备2021009384号-1anbei 冀公网安备13050302001966号