LangChain记忆（Memory）全解析：让你的应用更懂用户

# 定义一个名为BaseMemory的基础类class BaseMemory(Serializable, ABC):"""用于Chains中的内存的抽象基类。这里的内存指的是Chains中的状态。内存可以用来存储关于Chain的过去执行的信息，并将该信息注入到Chain的未来执行的输入中。例如，对于会话型Chains，内存可以用来存储会话，并自动将它们添加到未来的模型提示中，以便模型具有必要的上下文来连贯地响应最新的输入。"""
# 定义一个名为Config的子类class Config:"""为此pydantic对象配置。Pydantic是一个Python库，用于数据验证和设置管理，主要基于Python类型提示。"""# 允许在pydantic模型中使用任意类型。这通常用于允许复杂的数据类型。arbitrary_types_allowed = True# 下面是一些必须由子类实现的方法：# 定义一个属性，它是一个抽象方法。任何从BaseMemory派生的子类都需要实现此方法。# 此方法应返回该内存类将添加到链输入的字符串键。@property@abstractmethoddef memory_variables(self) -> List[str]:"""获取此内存类将添加到链输入的字符串键。"""# 定义一个抽象方法。任何从BaseMemory派生的子类都需要实现此方法。# 此方法基于给定的链输入返回键值对。@abstractmethoddef load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:"""根据给链的文本输入返回键值对。"""# 定义一个抽象方法。任何从BaseMemory派生的子类都需要实现此方法。# 此方法将此链运行的上下文保存到内存。@abstractmethoddef save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:"""保存此链运行的上下文到内存。"""# 定义一个抽象方法。任何从BaseMemory派生的子类都需要实现此方法。# 此方法清除内存内容。@abstractmethoddef clear(self) -> None:"""清除内存内容。"""

# 定义一个名为BaseChatMessageHistory的基础类class BaseChatMessageHistory(ABC):"""聊天消息历史记录的抽象基类。"""
# 在内存中存储的消息列表messages: List[BaseMessage]
# 定义一个add_user_message方法，它是一个方便的方法，用于将人类消息字符串添加到存储区。def add_user_message(self, message: str) -> None:"""为存储添加一个人类消息字符串的便捷方法。
参数:message: 人类消息的字符串内容。"""self.add_message(HumanMessage(content=message))
# 定义一个add_ai_message方法，它是一个方便的方法，用于将AI消息字符串添加到存储区。def add_ai_message(self, message: str) -> None:"""为存储添加一个AI消息字符串的便捷方法。
参数:message: AI消息的字符串内容。"""self.add_message(AIMessage(content=message))
# 抽象方法，需要由继承此基类的子类来实现。@abstractmethoddef add_message(self, message: BaseMessage) -> None:"""将Message对象添加到存储区。
参数:message: 要存储的BaseMessage对象。"""raise NotImplementedError()
# 抽象方法，需要由继承此基类的子类来实现。@abstractmethoddef clear(self) -> None:"""从存储中删除所有消息"""

def buffer_memory_test():"""使用ConversationBufferMemory测试对话流程，这个内存模型会存储所有对话历史，方便提取。"""conversation = ConversationChain(llm=llm, verbose=True, memory=ConversationBufferMemory() # ConversationBufferMemory可以用来存储消息，并将消息提取到一个变量中。)print(conversation.predict(input="你好呀！"))print(conversation.predict(input="你为什么叫小米？跟雷军有关系吗？"))

def buffer_window_memory_test():"""使用ConversationBufferWindowMemory测试对话流程，这个内存模型只保留最近几次交互，实现对话上下文的窗口式管理。"""conversation_with_summary = ConversationChain(llm=llm, # 我们设置一个较小的k值为2，只保留最后两次交互的记忆memory=ConversationBufferWindowMemory(k=2), verbose=True)print(conversation_with_summary.predict(input="嗨，你最近过得怎么样？"))print(conversation_with_summary.predict(input="你最近学到什么新知识了?"))print(conversation_with_summary.predict(input="展开讲讲？"))print(conversation_with_summary.predict(input="如果要构建聊天机器人，具体要用什么自然语言处理技术?"))

def summary_memory_test():"""使用ConversationSummaryBufferMemory测试对话流程，这个内存模型会摘要存储对话历史，且限定存储的token数量。"""memory = ConversationSummaryBufferMemory(llm=llm, max_token_limit=10)# 存储上下文信息，摘要形式保存对话memory.save_context({"input": "嗨，你最近过得怎么样？"}, {"output": " 嗨！我最近过得很好，谢谢你问。我最近一直在学习新的知识，并且正在尝试改进自己的性能。我也在尝试更多的交流，以便更好地了解人类的思维方式。"})memory.save_context({"input": "你最近学到什么新知识了?"}, {"output": " 最近我学习了有关自然语言处理的知识，以及如何更好地理解人类的语言。我还学习了有关机器学习的知识，以及如何使用它来改善自己的性能。"})print(memory.load_memory_variables({}))print(memory.load_memory_variables({})['history'])