如何添加消息历史记录

先决条件

本指南假定读者熟悉以下概念：

笔记

本指南之前介绍了 RunnableWithMessageHistory 抽象。您可以在 v0.2 文档中访问此版本的指南。

在 LangChain v0.3 版本发布后，我们建议 LangChain 用户利用 LangGraph 持久化将 memory 集成到新的 LangChain 应用中。

如果您的代码已经依赖于 RunnableWithMessageHistory 或 BaseChatMessageHistory，则无需进行任何更改。由于该功能适用于简单的聊天应用，且使用 RunnableWithMessageHistory 的任何代码将继续按预期工作，因此我们计划在近期不弃用此功能。

请参阅如何迁移到 LangGraph Memory 以获取更多详细信息。

在构建聊天机器人时，将对话状态传入和传出链至关重要。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.chat_models import init_chat_model

llm = init_chat_model("gpt-4o-mini", model_provider="openai")

示例：消息输入

为聊天模型添加记忆功能提供了一个简单的示例。聊天模型接收消息列表作为输入并输出一条消息。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)

API 参考：HumanMessage | MemorySaver | StateGraph

当运行应用程序时，我们传入一个配置 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

API 参考：ChatPromptTemplate | MessagesPlaceholder

在此场景中，我们定义图状态包含这些参数（除了消息历史记录之外）。然后，我们以与之前相同的方式定义一个单节点图。

请注意以下状态：

对 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()

==================================[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")]})

API 参考：HumanMessage

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 文档：

设置​

示例：消息输入​

示例：字典输入​

管理消息历史​

设置

示例：消息输入

示例：字典输入

管理消息历史