如何添加消息历史
本指南假设熟悉以下概念
本指南先前介绍了 RunnableWithMessageHistory 抽象。您可以在 v0.2 文档中访问本指南的此版本。
截至 LangChain 的 v0.3 版本,我们建议 LangChain 用户利用 LangGraph 持久性 将 memory
合并到新的 LangChain 应用程序中。
如果你的代码已经依赖 RunnableWithMessageHistory
或 BaseChatMessageHistory
,你无需进行任何更改。我们不打算在近期弃用此功能,因为它适用于简单的聊天应用程序,并且任何使用 RunnableWithMessageHistory
的代码都将继续按预期工作。
请参阅如何迁移到 LangGraph 内存以获取更多详细信息。
在构建聊天机器人时,将对话状态传入和传出链至关重要。LangGraph 实现了一个内置的持久层,允许链状态自动持久化在内存中,或外部后端(如 SQLite、Postgres 或 Redis)中。详细信息可以在 LangGraph 的持久化文档中找到。
在本指南中,我们将演示如何通过将任意 LangChain 可运行对象包装在一个最小的 LangGraph 应用程序中来添加持久化功能。这使我们能够持久化消息历史记录和链状态的其他元素,从而简化多轮应用程序的开发。它还支持多线程,使单个应用程序能够与多个用户分别交互。
设置
让我们初始化一个聊天模型
pip install -qU langchain-openai
import getpass
import os
if not os.environ.get("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
示例:消息输入
向聊天模型添加内存提供了一个简单的例子。聊天模型接受消息列表作为输入并输出消息。LangGraph 包括一个内置的 MessagesState
,我们可以将其用于此目的。
下面,我们
- 将图状态定义为消息列表;
- 向图中添加一个调用聊天模型的节点;
- 使用内存检查点器编译该图,以存储运行之间的消息。
LangGraph 应用程序的输出是其状态。这可以是任何 Python 类型,但在这种情况下,它通常是一个 TypedDict
,它与你的可运行对象的模式匹配。
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import START, MessagesState, StateGraph
# Define a new graph
workflow = StateGraph(state_schema=MessagesState)
# Define the function that calls the model
def call_model(state: MessagesState):
response = llm.invoke(state["messages"])
# Update message history with response:
return {"messages": response}
# Define the (single) node in the graph
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)
# Add memory
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
当我们运行应用程序时,我们会传入一个配置 dict
,其中指定了 thread_id
。此 ID 用于区分对话线程(例如,在不同用户之间)。
config = {"configurable": {"thread_id": "abc123"}}
然后我们可以调用该应用程序
query = "Hi! I'm Bob."
input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print() # output contains all messages in state
==================================[1m Ai Message [0m==================================
It's nice to meet you, Bob! I'm Claude, an AI assistant created by Anthropic. How can I help you today?
query = "What's my name?"
input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
Your name is Bob, as you introduced yourself at the beginning of our conversation.
请注意,不同线程的状态是分开的。如果我们向具有新 thread_id
的线程发出相同的查询,模型会指示它不知道答案
query = "What's my name?"
config = {"configurable": {"thread_id": "abc234"}}
input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
I'm afraid I don't actually know your name. As an AI assistant, I don't have personal information about you unless you provide it to me directly.
示例:字典输入
LangChain 可运行对象通常通过单个 dict
参数中的单独键接受多个输入。一个常见的例子是带有多个参数的提示模板。
之前我们的可运行对象是聊天模型,现在我们将提示模板和聊天模型链接在一起。
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages(
[
("system", "Answer in {language}."),
MessagesPlaceholder(variable_name="messages"),
]
)
runnable = prompt | llm
对于这种情况,我们定义图状态以包括这些参数(除了消息历史记录之外)。然后,我们以与之前相同的方式定义一个单节点图。
请注意,在下面的状态中
- 对
messages
列表的更新将附加消息; - 对
language
字符串的更新将覆盖该字符串。
from typing import Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from typing_extensions import Annotated, TypedDict
class State(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
language: str
workflow = StateGraph(state_schema=State)
def call_model(state: State):
response = runnable.invoke(state)
# Update message history with response:
return {"messages": [response]}
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "abc345"}}
input_dict = {
"messages": [HumanMessage("Hi, I'm Bob.")],
"language": "Spanish",
}
output = app.invoke(input_dict, config)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
¡Hola, Bob! Es un placer conocerte.
管理消息历史记录
可以通过 .get_state
访问消息历史记录(和应用程序状态的其他元素)
state = app.get_state(config).values
print(f'Language: {state["language"]}')
for message in state["messages"]:
message.pretty_print()
Language: Spanish
================================[1m Human Message [0m=================================
Hi, I'm Bob.
==================================[1m Ai Message [0m==================================
¡Hola, Bob! Es un placer conocerte.
我们还可以通过 .update_state
更新状态。例如,我们可以手动附加一条新消息
from langchain_core.messages import HumanMessage
_ = app.update_state(config, {"messages": [HumanMessage("Test")]})
state = app.get_state(config).values
print(f'Language: {state["language"]}')
for message in state["messages"]:
message.pretty_print()
Language: Spanish
================================[1m Human Message [0m=================================
Hi, I'm Bob.
==================================[1m Ai Message [0m==================================
¡Hola, Bob! Es un placer conocerte.
================================[1m Human Message [0m=================================
Test
有关管理状态的详细信息,包括删除消息,请参阅 LangGraph 文档