一、消息存储在内存
下面我们展示一个简单的示例，其中聊天历史保存在内存中，此处通过全局 Python 字典实现。
我们构建一个名为 get_session_history 的可调用对象，引用此字典以返回 ChatMessageHistory 实例。通过在运行时向 RunnableWithMessageHistory 传递配置，可以指定可调用对象的参数。默认情况下，期望配置参数是一个字符串 session_id。可以通过 history_factory_config 关键字参数进行调整。
使用单参数默认值：
1
2
3
4
5
6
7
8
9
10
11
12
|
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:if session_id not in store:
store[session_id] = ChatMessageHistory()return store[session_id]
with_message_history = RunnableWithMessageHistory(
runnable,
get_session_history,
input_messages_key="input",
history_messages_key="history",
)
|
请注意，我们已指定了 input_messages_key（要视为最新输入消息的键）和 history_messages_key（要添加历史消息的键）。
在调用此新 Runnable 时，我们通过配置参数指定相应的聊天历史：
1
2
3
4
| with_message_history.invoke(
{"ability": "math", "input": "余弦是什么意思？"},
config={"configurable": {"session_id": "abc123"}},
)
|
1
| content='余弦是一个数学函数，通常在三角学中使用，表示直角三角形的邻边和斜边的比例。' response_metadata={'token_usage': {'completion_tokens': 38, 'prompt_tokens': 38, 'total_tokens': 76}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-9aa23716-3959-476d-9386-6d433266e060-0' usage_metadata={'input_tokens': 38, 'output_tokens': 38, 'total_tokens': 76}
|
1
2
3
4
5
|
with_message_history.invoke(
{"ability": "math", "input": "什么?"},
config={"configurable": {"session_id": "abc123"}},
)
|
1
| content='余弦是一个数学术语，用于描述直角三角形中的角度关系。' response_metadata={'token_usage': {'completion_tokens': 26, 'prompt_tokens': 88, 'total_tokens': 114}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-f77baf90-6a13-4f48-991a-28e60ece84e8-0' usage_metadata={'input_tokens': 88, 'output_tokens': 26, 'total_tokens': 114}
|
1
2
3
4
5
|
with_message_history.invoke(
{"ability": "math", "input": "什么?"},
config={"configurable": {"session_id": "def234"}},
)
|
1
| content='对不起，我没明白你的问题。你能再详细一点吗？我很擅长数学。' response_metadata={'token_usage': {'completion_tokens': 34, 'prompt_tokens': 32, 'total_tokens': 66}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-3f69d281-a850-452f-8055-df70d4936630-0' usage_metadata={'input_tokens': 32, 'output_tokens': 34, 'total_tokens': 66}
|
二、配置会话唯一键
我们可以通过向 history_factory_config 参数传递一个 ConfigurableFieldSpec 对象列表来自定义跟踪消息历史的配置参数。下面我们使用了两个参数：user_id 和 conversation_id。
配置 user_id 和 conversation_id 作为会话唯一键：
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
29
30
| from langchain_core.runnables import ConfigurableFieldSpec
store = {}
def get_session_history(user_id: str, conversation_id: str) -> BaseChatMessageHistory:if (user_id, conversation_id) not in store:
store[(user_id, conversation_id)] = ChatMessageHistory()return store[(user_id, conversation_id)]
with_message_history = RunnableWithMessageHistory(
runnable,
get_session_history,
input_messages_key="input",
history_messages_key="history",
history_factory_config=[
ConfigurableFieldSpec(id="user_id",
annotation=str,
name="User ID",
description="用户的唯一标识符。",
default="",
is_shared=True,
),
ConfigurableFieldSpec(id="conversation_id",
annotation=str,
name="Conversation ID",
description="对话的唯一标识符。",
default="",
is_shared=True,
),
],
)
with_message_history.invoke(
{"ability": "math", "input": "余弦是什么意思？"},
config={"configurable": {"user_id": "123", "conversation_id": "1"}},
)
|
1
| content='对不起，你能提供一些更详细的信息吗？我会很高兴帮助你解决数学问题。' response_metadata={'token_usage': {'completion_tokens': 38, 'prompt_tokens': 32, 'total_tokens': 70}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-02030348-7bbb-4f76-8c68-61785d012c26-0' usage_metadata={'input_tokens': 32, 'output_tokens': 38, 'total_tokens': 70}
|
在许多情况下，持久化对话历史是可取的。RunnableWithMessageHistory 对于 get_session_history 可调用如何检索其聊天消息历史是中立的。请参见这里 ，这是一个使用本地文件系统的示例。下面我们演示如何使用 Redis。请查看内存集成页面，以获取使用其他提供程序的聊天消息历史的实现。
三、消息持久化
请查看 memory integrations 页面，了解使用 Redis 和其他提供程序实现聊天消息历史的方法。这里我们演示使用内存中的 ChatMessageHistory 以及使用 RedisChatMessageHistory 进行更持久存储。
1. 配置 Redis 环境
如果尚未安装 Redis，我们需要安装它：
1
| %pip install --upgrade --quiet redis
|
如果我们没有现有的 Redis 部署可以连接，可以启动本地 Redis Stack 服务器：
1
| docker run -d -p 6379:6379 -p 8001:8001 redis/redis-stack:latest
|
1
| REDIS_URL = "redis://localhost:6379/0"
|
2. 调用聊天接口，看 Redis 是否存储历史记录
更新消息历史实现只需要我们定义一个新的可调用对象，这次返回一个 RedisChatMessageHistory 实例：
1
2
3
4
5
6
7
8
| from langchain_community.chat_message_histories import RedisChatMessageHistory
def get_message_history(session_id: str) -> RedisChatMessageHistory:return RedisChatMessageHistory(session_id, url=REDIS_URL)
with_message_history = RunnableWithMessageHistory(
runnable,
get_message_history,
input_messages_key="input",
history_messages_key="history",
)
|
我们可以像以前一样调用：
1
2
3
4
| with_message_history.invoke(
{"ability": "math", "input": "余弦是什么意思？"},
config={"configurable": {"session_id": "foobar"}},
)
|
1
| content='余弦是一个三角函数，它表示直角三角形的邻边长度和斜边长度的比值。' response_metadata={'token_usage': {'completion_tokens': 33, 'prompt_tokens': 38, 'total_tokens': 71}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-2d1eba02-4709-4db5-ab6b-0fd03ab4c68a-0' usage_metadata={'input_tokens': 38, 'output_tokens': 33, 'total_tokens': 71}
|
1
2
3
4
| with_message_history.invoke(
{"ability": "math", "input": "什么?"},
config={"configurable": {"session_id": "foobar"}},
)
|
1
| content='余弦是一个数学术语，代表在一个角度下的邻边和斜边的比例。' response_metadata={'token_usage': {'completion_tokens': 32, 'prompt_tokens': 83, 'total_tokens': 115}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-99368d03-c2ed-4dda-a32f-677c036ad676-0' usage_metadata={'input_tokens': 83, 'output_tokens': 32, 'total_tokens': 115}
|
redis 历史记录查询：
四、修改聊天历史
修改存储的聊天消息可以帮助您的聊天机器人处理各种情况。以下是一些示例：
1. 裁剪消息
LLM 和聊天模型有限的上下文窗口，即使您没有直接达到限制，您可能也希望限制模型处理的干扰量。一种解决方案是只加载和存储最近的 n 条消息。让我们使用一个带有一些预加载消息的示例历史记录：
1
2
3
4
5
6
7
8
9
| temp_chat_history = ChatMessageHistory()
temp_chat_history.add_user_message("我叫Jack，你好")
temp_chat_history.add_ai_message("你好")
temp_chat_history.add_user_message("我今天心情挺开心")
temp_chat_history.add_ai_message("你今天心情怎么样")
temp_chat_history.add_user_message("我下午在打篮球")
temp_chat_history.add_ai_message("你下午在做什么")
temp_chat_history.messages
|
1
| [HumanMessage(content='我叫Jack，你好'), AIMessage(content='你好'), HumanMessage(content='我今天心情挺开心'), AIMessage(content='你今天心情怎么样'), HumanMessage(content='我下午在打篮球'), AIMessage(content='你下午在做什么'), HumanMessage(content='我今天心情如何?'), AIMessage(content='你今天的心情很开心。')]
|
让我们将这个消息历史与上面声明的 RunnableWithMessageHistory 链条一起使用：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| prompt = ChatPromptTemplate.from_messages(
[
("system","你是一个乐于助人的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的事实。",
),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
]
)
chain = prompt | chat
chain_with_message_history = RunnableWithMessageHistory(
chain,lambda session_id: temp_chat_history,
input_messages_key="input",
history_messages_key="chat_history",
)
chain_with_message_history.invoke(
{"input": "我今天心情如何?"},
{"configurable": {"session_id": "unused"}},
)
|
我们可以看到链条记住了预加载的名字。
但是假设我们有一个非常小的上下文窗口，并且我们想要将传递给链的消息数量减少到最近的2条。我们可以使用 clear 方法来删除消息并重新将它们添加到历史记录中。我们不一定要这样做，但让我们将这个方法放在链的最前面，以确保它总是被调用：
1
2
3
4
5
6
7
8
9
| from langchain_core.runnables import RunnablePassthrough
def trim_messages(chain_input):
stored_messages = temp_chat_history.messagesif len(stored_messages) <= 2:return False
temp_chat_history.clear()for message in stored_messages[-2:]:
temp_chat_history.add_message(message)return True
chain_with_trimming = (
RunnablePassthrough.assign(messages_trimmed=trim_messages)
| chain_with_message_history
)
|
让我们调用这个新链并检查消息：
1
2
3
4
| chain_with_trimming.invoke(
{"input": "我下午在做什么?"},
{"configurable": {"session_id": "unused"}},
)
|
1
| temp_chat_history.messages
|
1
| [HumanMessage(content='我下午在打篮球'), AIMessage(content='你下午在做什么'), HumanMessage(content='我下午在做什么?'), AIMessage(content='根据您之前的信息，您下午在打篮球。')]
|
我们可以看到我们的历史记录已经删除了两条最旧的消息，同时在末尾添加了最近的对话。下次调用链时，trim_messages 将再次被调用，只有最近的两条消息将被传递给模型。在这种情况下，这意味着下次调用时模型将忘记我们给它的名字：
1
2
3
4
| chain_with_trimming.invoke(
{"input": "我叫什么名字?"},
{"configurable": {"session_id": "unused"}},
)
|
1
| 对不起，我无法获取这个信息，因为你还没有告诉我你的名字。
|
1
| temp_chat_history.messages
|
1
| [HumanMessage(content='我下午在打篮球'), AIMessage(content='你下午在做什么'), HumanMessage(content='我叫什么名字?'), AIMessage(content='对不起，我无法获取这个信息，因为你还没有告诉我你的名字。')]
|
2. 总结记忆
我们也可以以其他方式使用相同的模式。例如，我们可以使用额外的 LLM 调用来在调用链之前生成对话摘要。让我们重新创建我们的聊天历史和聊天机器人链：
1
2
3
4
5
6
7
8
| temp_chat_history = ChatMessageHistory()
temp_chat_history.add_user_message("我叫Jack，你好")
temp_chat_history.add_ai_message("你好")
temp_chat_history.add_user_message("我今天心情挺开心")
temp_chat_history.add_ai_message("你今天心情怎么样")
temp_chat_history.add_user_message("我下午在打篮球")
temp_chat_history.add_ai_message("你下午在做什么")
temp_chat_history.messages
|
1
| [HumanMessage(content='我叫Jack，你好'), AIMessage(content='你好'), HumanMessage(content='我今天心情挺开心'), AIMessage(content='你今天心情怎么样'), HumanMessage(content='我下午在打篮球'), AIMessage(content='你下午在做什么'), HumanMessage(content='我今天心情如何?'), AIMessage(content='作为一个人工智能，我无法知道你的心情。你可以告诉我你今天感觉如何，我会尽我所能提供帮助。')]
|
我们将稍微修改提示，让 LLM 意识到它将收到一个简短摘要而不是聊天历史：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| prompt = ChatPromptTemplate.from_messages(
[
("system","你是一个乐于助人的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的事实。",
),
MessagesPlaceholder(variable_name="chat_history"),
("user", "{input}"),
]
)
chain = prompt | chat
chain_with_message_history = RunnableWithMessageHistory(
chain,lambda session_id: temp_chat_history,
input_messages_key="input",
history_messages_key="chat_history",
)
|
现在，让我们创建一个函数，将之前的交互总结为摘要。我们也可以将这个函数添加到链的最前面：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| def summarize_messages(chain_input):
stored_messages = temp_chat_history.messagesif len(stored_messages) == 0:return False
summarization_prompt = ChatPromptTemplate.from_messages(
[
MessagesPlaceholder(variable_name="chat_history"),
("user","将上述聊天消息浓缩成一条摘要消息。尽可能包含多个具体细节。",
),
]
)
summarization_chain = summarization_prompt | chat
summary_message = summarization_chain.invoke({"chat_history": stored_messages})
temp_chat_history.clear()
temp_chat_history.add_message(summary_message)return True
chain_with_summarization = (
RunnablePassthrough.assign(messages_summarized=summarize_messages)
| chain_with_message_history
)
|
让我们看看它是否记得我们给它起的名字：
1
2
3
4
| chain_with_summarization.invoke(
{"input": "我下午在干嘛"},
{"configurable": {"session_id": "unused"}},
)
|
输出结果为：
查看聊天历史记录：
1
| temp_chat_history.messages
|
输出结果为：
1
| [AIMessage(content='用户Jack今天心情很好，他下午打了篮球。', response_metadata={'token_usage': {'completion_tokens': 20, 'prompt_tokens': 108, 'total_tokens': 128}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-3ece2bde-b763-4ca0-84f9-43cfdf5c2e5e-0', usage_metadata={'input_tokens': 108, 'output_tokens': 20, 'total_tokens': 128}), HumanMessage(content='我下午在干嘛'), AIMessage(content='下午你在打篮球。'})]
|
请注意，再次调用链式模型会生成一个新的摘要，该摘要包括初始摘要以及新的消息等。您还可以设计一种混合方法，其中一定数量的消息保留在聊天历史记录中，而其他消息则被摘要。
