`langchain_core.chat_history`.BaseChatMessageHistory¶

class langchain_core.chat_history.BaseChatMessageHistory[source]¶

存储聊天历史记录的抽象基类。

实现指南

预期实现将覆盖以下所有或部分方法

add_messages：添加消息的同步批次变体
aadd_messages：添加消息的异步批次变体
messages：获取消息的同步批次变体
aget_messages：获取消息的异步批次变体
clear：清除消息的同步批次变体
aclear：清除消息的异步批次变体

add_messages包含一个默认实现，该实现会为序列中的每条消息调用add_message。这是为了向后兼容现有的实现，其中只包含了add_message。

所有异步变体都提供了默认实现，该实现会调用同步变体。实现者可以选择覆盖异步实现以提供真正的异步实现。

使用指南

当用于更新历史记录时，用户应优先使用add_messages而不是add_message或其他的变体，如add_user_message和add_ai_message，以避免不必要的底层持久层往返。

示例：显示默认实现。

class FileChatMessageHistory(BaseChatMessageHistory):
    storage_path:  str
    session_id: str

   @property
   def messages(self):
       with open(os.path.join(storage_path, session_id), 'r:utf-8') as f:
           messages = json.loads(f.read())
        return messages_from_dict(messages)

   def add_messages(self, messages: Sequence[BaseMessage]) -> None:
       all_messages = list(self.messages) # Existing messages
       all_messages.extend(messages) # Add new messages

       serialized = [message_to_dict(message) for message in all_messages]
       # Can be further optimized by only writing new messages
       # using append mode.
       with open(os.path.join(storage_path, session_id), 'w') as f:
           json.dump(f, messages)

   def clear(self):
       with open(os.path.join(storage_path, session_id), 'w') as f:
           f.write("[]")

属性

messages

一个属性或属性，返回消息列表。

方法

`__init__`()
`aadd_messages`(messages)	异步添加多条消息。
`aclear`()	异步从存储中删除所有消息
`add_ai_message`(message)	方便函数，将AI消息字符串添加到存储中。
`add_message`(message)	将消息对象添加到存储。
`add_messages`(messages)	添加消息列表。
`add_user_message`(message)	方便函数，将人类消息字符串添加到存储。
`aget_messages`()	获取消息的异步版本。
`clear`()	从存储中删除所有消息

__init__()¶

async aadd_messages(messages: Sequence[BaseMessage]) → None[source]¶

异步添加多条消息。

参数: messages (Sequence[BaseMessage]) – 要存储的BaseMessage对象序列。
返回类型: None

async aclear() → None[源代码]¶

异步从存储中删除所有消息

返回类型: None

add_ai_message(message: Union[AIMessage, str]) → None[源代码]¶

方便函数，将AI消息字符串添加到存储中。

请注意，这是一个便利方法。代码应优先使用批量add_messages接口，以节省底层持久化层的往返。

该方法可能在未来的版本中被弃用。

参数: 消息（联合： AIMessage ， str） – 要添加的AI消息。
返回类型: None

add_message(message: BaseMessage) → None[源代码]¶

将消息对象添加到存储。

参数: 消息（BaseMessage） – 要存储的BaseMessage对象。
引发的异常: NotImplementedError – 如果子类未实现高效的add_messages方法。
返回类型: None

add_messages(messages: Sequence[BaseMessage]) → None[源代码]¶

添加消息列表。

实现应覆盖此方法以以高效的方式处理消息的批量添加，以避免不必要的对底层存储的往返。

参数: messages (Sequence[BaseMessage]) – 要存储的BaseMessage对象序列。
返回类型: None

add_user_message(message: Union[HumanMessage, str]) → None[source]¶

方便函数，将人类消息字符串添加到存储。

请注意，这是一个便利方法。代码应优先使用批量add_messages接口，以节省底层持久化层的往返。

该方法可能在未来的版本中被弃用。

参数: message (Union[HumanMessage, str]) – 添加到存储中的人的消息。
返回类型: None

async aget_messages() → List[BaseMessage][source]¶

获取消息的异步版本。

可以重写此方法以提供高效的异步实现。

通常，获取消息可能涉及到底层持久化层的IO。

返回类型: List[BaseMessage]

abstract clear() → None[source]¶

从存储中删除所有消息

返回类型: None

使用 BaseChatMessageHistory 的示例¶

langchain_core.chat_history.BaseChatMessageHistory¶

使用 BaseChatMessageHistory 的示例¶

`langchain_core.chat_history`.BaseChatMessageHistory¶