跳到主要内容
Open In ColabOpen on GitHub

如何添加消息历史

先决条件
注意

本指南以前介绍过 RunnableWithMessageHistory 抽象。您可以在 v0.2 文档中查看本指南的此版本。

自 LangChain v0.3 版本发布以来,我们建议 LangChain 用户利用 LangGraph 持久化功能,将 memory 集成到新的 LangChain 应用程序中。

如果您的代码已经依赖于 RunnableWithMessageHistoryBaseChatMessageHistory,则无需进行任何更改。我们近期不打算弃用此功能,因为它适用于简单的聊天应用程序,并且任何使用 RunnableWithMessageHistory 的代码将继续按预期工作。

请参阅如何迁移到LangGraph记忆了解更多详细信息。

在构建聊天机器人时,将对话状态传入和传出链是至关重要的。LangGraph 实现了一个内置的持久化层,允许链状态自动持久化到内存或外部后端,例如 SQLite、Postgres 或 Redis。详细信息可在 LangGraph 持久化文档中找到。

在本指南中,我们将演示如何通过将任意 LangChain runnable 封装到最小化的 LangGraph 应用程序中,为其添加持久化功能。这使我们能够持久化消息历史和链状态的其他元素,从而简化多轮应用程序的开发。它还支持多线程,使单个应用程序能够与多个用户独立交互。

设置

让我们初始化一个聊天模型

pip install -qU "langchain[google-genai]"
import getpass
import os

if not os.environ.get("GOOGLE_API_KEY"):
os.environ["GOOGLE_API_KEY"] = getpass.getpass("Enter API key for Google Gemini: ")

from langchain.chat_models import init_chat_model

llm = init_chat_model("gemini-2.0-flash", model_provider="google_genai")

示例:消息输入

聊天模型添加内存提供了一个简单的示例。聊天模型接受消息列表作为输入并输出一条消息。LangGraph 包含一个内置的 MessagesState,我们可以为此目的使用它。

下面,我们

  1. 将图状态定义为消息列表;
  2. 向图中添加一个调用聊天模型的节点;
  3. 使用内存中的检查点(checkpointer)编译图,以在运行之间存储消息。
信息

LangGraph 应用程序的输出是其 状态。这可以是任何 Python 类型,但在本上下文中,它通常是一个与您的 runnable 的模式匹配的 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
================================== Ai Message ==================================

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()
================================== Ai Message ==================================

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()
================================== Ai Message ==================================

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 runnables 通常通过单个 dict 参数中的不同键接受多个输入。一个常见示例是具有多个参数的提示模板。

之前我们的 runnable 是一个聊天模型,而在这里我们则将提示模板和聊天模型链接在一起。

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)
API 参考:BaseMessage | add_messages
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()
================================== Ai Message ==================================

¡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
================================ Human Message =================================

Hi, I'm Bob.
================================== Ai Message ==================================

¡Hola, Bob! Es un placer conocerte.

我们还可以通过 .update_state 更新状态。例如,我们可以手动添加一条新消息

from langchain_core.messages import HumanMessage

_ = app.update_state(config, {"messages": [HumanMessage("Test")]})
API 参考:HumanMessage
state = app.get_state(config).values

print(f"Language: {state['language']}")
for message in state["messages"]:
message.pretty_print()
Language: Spanish
================================ Human Message =================================

Hi, I'm Bob.
================================== Ai Message ==================================

¡Hola, Bob! Es un placer conocerte.
================================ Human Message =================================

Test

有关管理状态(包括删除消息)的详细信息,请参阅 LangGraph 文档