大模型从入门到应用——LangChain：记忆（Memory）-[聊天消息记录]

分类目录：《大模型从入门到应用》总目录

LangChain系列文章：

• 基础知识
• 快速入门
  • 安装与环境配置
  • 链（Chains）、代理（Agent:）和记忆（Memory）
  • 快速开发聊天模型
• 模型（Models）
  • 基础知识
  • 大型语言模型（LLMs）
    • 基础知识
    • LLM的异步API、自定义LLM包装器、虚假LLM和人类输入LLM（Human Input LLM）
    • 缓存LLM的调用结果
    • 加载与保存LLM类、流式传输LLM与Chat Model响应和跟踪tokens使用情况
  • 聊天模型（Chat Models）
    • 基础知识
    • 使用少量示例和响应流式传输
  • 文本嵌入模型
    • Aleph Alpha、Amazon Bedrock、Azure OpenAI、Cohere等
    • Embaas、Fake Embeddings、Google Vertex AI PaLM等
• 提示（Prompts）
  • 基础知识
  • 提示模板
    • 基础知识
    • 连接到特征存储
    • 创建自定义提示模板和含有Few-Shot示例的提示模板
    • 部分填充的提示模板和提示合成
    • 序列化提示信息
  • 示例选择器（Example Selectors）
  • 输出解析器（Output Parsers）
• 记忆（Memory）
  • 基础知识
  • 记忆的类型
    • 会话缓存记忆、会话缓存窗口记忆和实体记忆
    • 对话知识图谱记忆、对话摘要记忆和会话摘要缓冲记忆
    • 对话令牌缓冲存储器和基于向量存储的记忆
  • 将记忆添加到LangChain组件中
  • 自定义对话记忆与自定义记忆类
  • 聊天消息记录
  • 记忆的存储与应用
• 索引（Indexes）
  • 基础知识
  • 文档加载器（Document Loaders）
  • 文本分割器（Text Splitters）
  • 向量存储器（Vectorstores）
  • 检索器（Retrievers）
• 链（Chains）
  • 基础知识
  • 通用功能
    • 自定义Chain和Chain的异步API
    • LLMChain和RouterChain
    • SequentialChain和TransformationChain
    • 链的保存（序列化）与加载（反序列化）
  • 链与索引
    • 文档分析和基于文档的聊天
    • 问答的基础知识
    • 图问答（Graph QA）和带来源的问答（Q&A with Sources）
    • 检索式问答
    • 文本摘要（Summarization）、HyDE和向量数据库的文本生成
• 代理（Agents）
  • 基础知识
  • 代理类型
  • 自定义代理（Custom Agent）
  • 自定义MRKL代理
  • 带有ChatModel的LLM聊天自定义代理和自定义多操作代理（Custom MultiAction Agent）
  • 工具
    • 基础知识
    • 自定义工具（Custom Tools）
    • 多输入工具和工具输入模式
    • 人工确认工具验证和Tools作为OpenAI函数
  • 工具包（Toolkit）
  • 代理执行器（Agent Executor）
    • 结合使用Agent和VectorStore
    • 使用Agents的异步API和创建ChatGPT克隆
    • 处理解析错误、访问中间步骤和限制最大迭代次数
    • 为代理程序设置超时时间和限制最大迭代次数和为代理程序和其工具添加共享内存
  • 计划与执行
• 回调函数（Callbacks）

---

Cassandra聊天消息记录

Cassandra是一种分布式数据库，非常适合存储大量数据，是存储聊天消息历史的良好选择，因为它易于扩展，能够处理大量写入操作。

# List of contact points to try connecting to Cassandra cluster.
contact_points = ["cassandra"]

from langchain.memory import CassandraChatMessageHistory

message_history = CassandraChatMessageHistory(
    contact_points=contact_points, session_id="test-session"
)

message_history.add_user_message("hi!")

message_history.add_ai_message("whats up?")
message_history.messages
[HumanMessage(content='hi!', additional_kwargs={}, example=False),
AIMessage(content='whats up?', additional_kwargs={}, example=False)]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

DynamoDB聊天消息记录

首先确保我们已经正确配置了AWS CLI，并再确保我们已经安装了boto3。接下来，创建我们将存储消息 DynamoDB表：

import boto3

# Get the service resource.
dynamodb = boto3.resource('dynamodb')

# Create the DynamoDB table.
table = dynamodb.create_table(
    TableName='SessionTable',
    KeySchema=[
        {
            'AttributeName': 'SessionId',
            'KeyType': 'HASH'
        }
    ],
    AttributeDefinitions=[
        {
            'AttributeName': 'SessionId',
            'AttributeType': 'S'
        }
    ],
    BillingMode='PAY_PER_REQUEST',
)

# Wait until the table exists.
table.meta.client.get_waiter('table_exists').wait(TableName='SessionTable')

# Print out some data about the table.
print(table.item_count)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

输出：

0
1

DynamoDBChatMessageHistory

from langchain.memory.chat_message_histories import DynamoDBChatMessageHistory

history = DynamoDBChatMessageHistory(table_name="SessionTable", session_id="0")
history.add_user_message("hi!")
history.add_ai_message("whats up?")
history.messages
1
2
3
4
5
6

输出：

[HumanMessage(content='hi!', additional_kwargs={}, example=False),
AIMessage(content='whats up?', additional_kwargs={}, example=False)]
1
2

使用自定义端点URL的DynamoDBChatMessageHistory

有时候在连接到AWS端点时指定URL非常有用，比如在本地使用Localstack进行开发。对于这种情况，我们可以通过构造函数中的endpoint_url参数来指定URL。

from langchain.memory.chat_message_histories import DynamoDBChatMessageHistory

history = DynamoDBChatMessageHistory(table_name="SessionTable", session_id="0", endpoint_url="http://localhost.localstack.cloud:4566")
1
2
3

Agent with DynamoDB Memory

from langchain.agents import Tool
from langchain.memory import ConversationBufferMemory
from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent
from langchain.agents import AgentType
from langchain.utilities import PythonREPL
from getpass import getpass

message_history = DynamoDBChatMessageHistory(table_name="SessionTable", session_id="1")
memory = ConversationBufferMemory(memory_key="chat_history", chat_memory=message_history, return_messages=True)
python_repl = PythonREPL()

# You can create the tool to pass to an agent
tools = [Tool(
    name="python_repl",
    description="A Python shell. Use this to execute python commands. Input should be a valid python command. If you want to see the output of a value, you should print it out with `print(...)`.",
    func=python_repl.run
)]
llm=ChatOpenAI(temperature=0)
agent_chain = initialize_agent(tools, llm, agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, verbose=True, memory=memory)
agent_chain.run(input="Hello!")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

日志输出：

> Entering new AgentExecutor chain...
{
    "action": "Final Answer",
    "action_input": "Hello! How can I assist you today?"
}

> Finished chain.
1
2
3
4
5
6
7

输出：

'Hello! How can I assist you today?'
1

输入：

agent_chain.run(input="Who owns Twitter?")
1

日志输出：

> Entering new AgentExecutor chain...
{
    "action": "python_repl",
    "action_input": "import requests\nfrom bs4 import BeautifulSoup\n\nurl = 'https://en.wikipedia.org/wiki/Twitter'\nresponse = requests.get(url)\nsoup = BeautifulSoup(response.content, 'html.parser')\nowner = soup.find('th', text='Owner').find_next_sibling('td').text.strip()\nprint(owner)"
}
Observation: X Corp. (2023–present)Twitter, Inc. (2006–2023)

Thought:{
    "action": "Final Answer",
    "action_input": "X Corp. (2023–present)Twitter, Inc. (2006–2023)"
}

> Finished chain.
1
2
3
4
5
6
7
8
9
10
11
12
13

输出：

'X Corp. (2023–present)Twitter, Inc. (2006–2023)'
1

输入：

agent_chain.run(input="My name is Bob.")
1

日志输出：

> Entering new AgentExecutor chain...
{
    "action": "Final Answer",
    "action_input": "Hello Bob! How can I assist you today?"
}

> Finished chain.
1
2
3
4
5
6
7

输出：

  'Hello Bob! How can I assist you today?'
1

输入：

agent_chain.run(input="Who am I?")
1

日志输出：

> Entering new AgentExecutor chain...
{
    "action": "Final Answer",
    "action_input": "Your name is Bob."
}

> Finished chain.
1
2
3
4
5
6
7

输出：

'Your name is Bob.'
1

Momento聊天消息记录

本节介绍如何使用Momento Cache来存储聊天消息记录，我们会使用MomentoChatMessageHistory类。需要注意的是，默认情况下，如果不存在具有给定名称的缓存，我们将创建一个新的缓存。我们需要获得一个Momento授权令牌才能使用这个类。这可以直接通过将其传递给momento.CacheClient实例化，作为MomentoChatMessageHistory.from_client_params的命名参数auth_token，或者可以将其设置为环境变量MOMENTO_AUTH_TOKEN。

from datetime import timedelta
from langchain.memory import MomentoChatMessageHistory

session_id = "foo"
cache_name = "langchain"
ttl = timedelta(days=1)
history = MomentoChatMessageHistory.from_client_params(
    session_id, 
    cache_name,
    ttl,
)

history.add_user_message("hi!")

history.add_ai_message("whats up?")
history.messages
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

输出：

[HumanMessage(content='hi!', additional_kwargs={}, example=False),
AIMessage(content='whats up?', additional_kwargs={}, example=False)]
1
2

MongoDB聊天消息记录

本节介绍如何使用MongoDB存储聊天消息记录。MongoDB是一个开放源代码的跨平台文档导向数据库程序。它被归类为NoSQL数据库程序，使用类似JSON的文档，并且支持可选的模式。MongoDB由MongoDB Inc.开发，并在服务器端公共许可证（SSPL）下许可。

# Provide the connection string to connect to the MongoDB database
connection_string = "mongodb://mongo_user:password123@mongo:27017"
from langchain.memory import MongoDBChatMessageHistory

message_history = MongoDBChatMessageHistory(
        connection_string=connection_string, session_id="test-session"
    )

message_history.add_user_message("hi!")

message_history.add_ai_message("whats up?")
message_history.messages
1
2
3
4
5
6
7
8
9
10
11
12

输出：

[HumanMessage(content='hi!', additional_kwargs={}, example=False),
AIMessage(content='whats up?', additional_kwargs={}, example=False)]
1
2

Postgres聊天消息历史记录

本节介绍了如何使用 Postgres 来存储聊天消息历史记录。

from langchain.memory import PostgresChatMessageHistory

history = PostgresChatMessageHistory(connection_string="postgresql://postgres:mypassword@localhost/chat_history", session_id="foo")

history.add_user_message("hi!")

history.add_ai_message("whats up?")
history.messages
1
2
3
4
5
6
7
8

Redis聊天消息历史记录

本节介绍了如何使用Redis来存储聊天消息历史记录。

from langchain.memory import RedisChatMessageHistory

history = RedisChatMessageHistory("foo")

history.add_user_message("hi!")
history.add_ai_message("whats up?")
history.messages
1
2
3
4
5
6
7

输出：

[AIMessage(content='whats up?', additional_kwargs={}),
HumanMessage(content='hi!', additional_kwargs={})]
1
2

参考文献：
[1] LangChain官方网站：https://www.langchain.com/
[2] LangChain