如何新增訊息歷史記錄

先決條件

本指南假設您熟悉以下概念

注意

本指南先前涵蓋了 RunnableWithMessageHistory 抽象概念。您可以在 v0.2 文件中存取本指南的此版本。

從 LangChain v0.3 版本開始，我們建議 LangChain 使用者利用 LangGraph 持久性將 memory 整合到新的 LangChain 應用程式中。

如果您的程式碼已經依賴 RunnableWithMessageHistory 或 BaseChatMessageHistory，則您不需要進行任何變更。我們不打算在近期內棄用此功能，因為它適用於簡單的聊天應用程式，且任何使用 RunnableWithMessageHistory 的程式碼都將繼續如預期般運作。

請參閱如何遷移到 LangGraph 記憶體以取得更多詳細資訊。

在建構聊天機器人時，將對話狀態傳遞到鏈中和鏈外至關重要。LangGraph 實作了內建持久性層，允許鏈狀態自動持久儲存在記憶體或外部後端（例如 SQLite、Postgres 或 Redis）中。詳細資訊可以在 LangGraph 持久性文件中找到。

在本指南中，我們示範如何透過將任意 LangChain runnable 包裝在最小 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 類型，但在這種情況下，它通常會是符合您 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)

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 runnable 通常透過單一 dict 引數中的不同鍵接受多個輸入。一個常見的範例是具有多個參數的提示範本。

之前我們的 runnable 是聊天模型，而這裡我們將提示範本和聊天模型鏈接在一起。

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 文件

設定​

範例：訊息輸入​

範例：字典輸入​

管理訊息歷史記錄​

此頁面是否對您有幫助？

設定

範例：訊息輸入

範例：字典輸入

管理訊息歷史記錄