添加消息历史记录（内存）

RunnableWithMessageHistory 允许我们将消息历史记录添加到某些类型的链中。它包装另一个 Runnable 并管理它的聊天消息历史记录。

具体来说，它可用于任何将以下之一作为输入的 Runnable

BaseMessage 序列
一个字典，其键采用 BaseMessage 序列
带有一个键的字典，该键将最新消息作为 BaseMessage 的字符串或序列，以及一个单独的键，该键将历史消息

并作为输出之一返回

一个字符串可以视为 AIMessage 的内容
BaseMessage 序列
一个字典，其键包含 BaseMessage 序列

让我们看一些示例，看看它是如何工作的。首先我们构造一个可运行的（这里接受一个字典作为输入并返回一条消息作为输出）：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai.chat_models import ChatOpenAI

model = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You're an assistant who's good at {ability}. Respond in 20 words or fewer",
        ),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)
runnable = prompt | model

为了管理消息历史记录，我们需要： 1. 这个可运行程序； 2. 返回 BaseChatMessageHistory 实例的可调用函数。

查看内存集成页面，了解使用 Redis 和其他提供程序实现聊天消息历史记录。在这里，我们演示如何使用内存中的 ChatMessageHistory 以及使用 RedisChatMessageHistory 的更持久的存储。

在内存中

下面我们展示了一个简单的示例，其中聊天历史记录位于内存中，在本例中通过全局 Python 字典。

我们构造一个可调用的 get_session_history 来引用此字典以返回 ChatMessageHistory 的实例。可以通过在运行时将配置传递给 RunnableWithMessageHistory 来指定可调用的参数。默认情况下，配置参数应为单个字符串 session_id。这可以通过history_factory_config kwarg进行调整。

使用单参数默认值：

from langchain_community.chat_message_histories import ChatMessageHistory
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 （添加历史消息的键）。

当调用这个新的可运行时，我们通过配置参数指定相应的聊天历史记录：

with_message_history.invoke(
    {"ability": "math", "input": "What does cosine mean?"},
    config={"configurable": {"session_id": "abc123"}},
)

AIMessage(content='Cosine is a trigonometric function that calculates the ratio of the adjacent side to the hypotenuse of a right triangle.')

# Remembers
with_message_history.invoke(
    {"ability": "math", "input": "What?"},
    config={"configurable": {"session_id": "abc123"}},
)

AIMessage(content='Cosine is a mathematical function used to calculate the length of a side in a right triangle.')

# New session_id --> does not remember.
with_message_history.invoke(
    {"ability": "math", "input": "What?"},
    config={"configurable": {"session_id": "def234"}},
)

AIMessage(content='I can help with math problems. What do you need assistance with?')

我们可以通过将 ConfigurableFieldSpec 对象列表传递给history_factory_config 参数来自定义用于跟踪消息历史记录的配置参数。下面，我们使用两个参数：user_id 和conversation_id。

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="Unique identifier for the user.",
            default="",
            is_shared=True,
        ),
        ConfigurableFieldSpec(
            id="conversation_id",
            annotation=str,
            name="Conversation ID",
            description="Unique identifier for the conversation.",
            default="",
            is_shared=True,
        ),
    ],
)

with_message_history.invoke(
    {"ability": "math", "input": "Hello"},
    config={"configurable": {"user_id": "123", "conversation_id": "1"}},
)

具有不同签名的可运行程序的示例

上面的可运行程序接受一个字典作为输入并返回一个 BaseMessage。下面我们展示了一些替代方案。

消息输入，字典输出

from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableParallel

chain = RunnableParallel({"output_message": ChatOpenAI()})


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(
    chain,
    get_session_history,
    output_messages_key="output_message",
)

with_message_history.invoke(
    [HumanMessage(content="What did Simone de Beauvoir believe about free will")],
    config={"configurable": {"session_id": "baz"}},
)

{'output_message': AIMessage(content="Simone de Beauvoir believed in the existence of free will. She argued that individuals have the ability to make choices and determine their own actions, even in the face of social and cultural constraints. She rejected the idea that individuals are purely products of their environment or predetermined by biology or destiny. Instead, she emphasized the importance of personal responsibility and the need for individuals to actively engage in creating their own lives and defining their own existence. De Beauvoir believed that freedom and agency come from recognizing one's own freedom and actively exercising it in the pursuit of personal and collective liberation.")}

with_message_history.invoke(
    [HumanMessage(content="How did this compare to Sartre")],
    config={"configurable": {"session_id": "baz"}},
)

{'output_message': AIMessage(content='Simone de Beauvoir\'s views on free will were closely aligned with those of her contemporary and partner Jean-Paul Sartre. Both de Beauvoir and Sartre were existentialist philosophers who emphasized the importance of individual freedom and the rejection of determinism. They believed that human beings have the capacity to transcend their circumstances and create their own meaning and values.\n\nSartre, in his famous work "Being and Nothingness," argued that human beings are condemned to be free, meaning that we are burdened with the responsibility of making choices and defining ourselves in a world that lacks inherent meaning. Like de Beauvoir, Sartre believed that individuals have the ability to exercise their freedom and make choices in the face of external and internal constraints.\n\nWhile there may be some nuanced differences in their philosophical writings, overall, de Beauvoir and Sartre shared a similar belief in the existence of free will and the importance of individual agency in shaping one\'s own life.')}

消息输入、消息输出

RunnableWithMessageHistory(
    ChatOpenAI(),
    get_session_history,
)

单键字典可用于所有消息输入、消息输出

from operator import itemgetter

RunnableWithMessageHistory(
    itemgetter("input_messages") | ChatOpenAI(),
    get_session_history,
    input_messages_key="input_messages",
)

持久存储

在许多情况下，最好保留对话历史记录。 RunnableWithMessageHistory 不知道 get_session_history 可调用函数如何检索其聊天消息历史记录。有关使用本地文件系统的示例，请参阅此处。下面我们演示如何使用 Redis。查看内存集成页面，了解使用其他提供程序实现聊天消息历史记录。

设置

如果尚未安装 Redis，我们需要安装它：

%pip install --upgrade --quiet redis

如果我们没有可连接的现有 Redis 部署，请启动本地 Redis Stack 服务器：

docker run -d -p 6379:6379 -p 8001:8001 redis/redis-stack:latest

REDIS_URL = "redis://localhost:6379/0"

LangSmith

LangSmith 对于诸如消息历史注入之类的事情特别有用，否则很难理解链各个部分的输入是什么。

请注意，LangSmith 不是必需的，但它很有帮助。如果您确实想使用 LangSmith，在上面的链接注册后，请确保取消注释以下内容并设置环境变量以开始记录跟踪：

# os.environ["LANGCHAIN_TRACING_V2"] = "true"
# os.environ["LANGCHAIN_API_KEY"] = getpass.getpass()

更新消息历史记录实现只需要我们定义一个新的可调用对象，这次返回 RedisChatMessageHistory 的实例：

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",
)

我们可以像以前一样调用：

with_message_history.invoke(
    {"ability": "math", "input": "What does cosine mean?"},
    config={"configurable": {"session_id": "foobar"}},
)

AIMessage(content='Cosine is a trigonometric function that represents the ratio of the adjacent side to the hypotenuse in a right triangle.')

with_message_history.invoke(
    {"ability": "math", "input": "What's its inverse"},
    config={"configurable": {"session_id": "foobar"}},
)

AIMessage(content='The inverse of cosine is the arccosine function, denoted as acos or cos^-1, which gives the angle corresponding to a given cosine value.')

Langsmith 跟踪

查看第二次调用的 Langsmith 跟踪，我们可以看到在构造提示时，已注入一个“历史”变量，它是两个消息的列表（我们的第一个输入和第一个输出）。

在内存中​

具有不同签名的可运行程序的示例​

持久存储​

设置​

LangSmith​

在内存中

具有不同签名的可运行程序的示例

持久存储

设置

LangSmith