1. Memory组件

在 LangChain 中，大多数与记忆相关的功能都被标记为 beta，无论是 0.1 还是 0.2 的版本，这有两个原因：

  1. 大多数功能（有些例外，见下文）尚未准备好投入生产。
  2. 大多数功能（有些例外，见下文）适用于传统链，而不是较新的 LCEL 语法。
    例外的案例是 ChatMessageHistory 功能，
    Memory 组件的基类是 BaseMemory，封装了大量的基础方法，例如：memory_variables、load_memory_variables、aload_memory_variables、save_context、asave_context、clear、aclear 函数。
    基于 BaseMemory 基类，衍生出了两个子类 SimpleMemory 和 BaseChatMemory，当 LLM 应用不需要记忆功能，又不想更换代码结构时，可以将记忆组件使用 SimpleMemory 组件进行代替，SimpleMemory 实现了记忆组件的相关方法，但是不存储任何记忆，可以在不修改代码结构的情况下替换记忆组件，实现无记忆功能。
    而 BaseChatMemory 组件是 LangChain 中内置的其他记忆组件的基类，针对对话历史进行了特定的封装，以适用聊天模型对话的场合。
    LangChain 记忆组件的流程图如下

2.BaseChatMemory 运行流程及源码

在 LangChain 的 BaseChatMemory 组件中，不同的属性与方法有不同的作用：

  1. chat_memory：用于管理记忆中的历史消息对话。
  2. output_key：定义 AI 内容输出键。
  3. input_key：定义 Human 内容输入键。
  4. return_messages：load_memory_variables 函数是否返回消息列表，默认为 False 代表返回字符串。
  5. save_context：存储上下文到记忆组件中（存储消息对话）。
  6. load_memory_variables：生成加载到链的记忆字典信息。
  7. clear：清除记忆中的对话消息历史。
    在聊天机器人的运行流程中，添加 BaseChatMemory 组件后，整体流程变化如下：
class BaseChatMemory(BaseMemory, ABC):chat_memory: BaseChatMessageHistory = Field(default_factory=InMemoryChatMessageHistory)output_key: Optional[str] = Noneinput_key: Optional[str] = Nonereturn_messages: bool = Falsedef _get_input_output(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> Tuple[str, str]:"""从输入和输出字典中提取对应的字符串（人类提问、AI输出）"""if self.input_key is None:# 如果没有传递input_key则从inputs中提取人类提问的keyprompt_input_key = get_prompt_input_key(inputs, self.memory_variables)else:prompt_input_key = self.input_keyif self.output_key is None:if len(outputs) == 1:output_key = list(outputs.keys())[0]elif "output" in outputs:output_key = "output"warnings.warn(f"'{self.__class__.__name__}' got multiple output keys:"f" {outputs.keys()}. The default 'output' key is being used."f" If this is not desired, please manually set 'output_key'.")else:raise ValueError(f"Got multiple output keys: {outputs.keys()}, cannot "f"determine which to store in memory. Please set the "f"'output_key' explicitly.")else:output_key = self.output_keyreturn inputs[prompt_input_key], outputs[output_key]def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:"""保存对话上下文到记忆缓冲区"""# 从输入和输出中获取输入字符串、输出字符串input_str, output_str = self._get_input_output(inputs, outputs)# 存储对应的对话信息self.chat_memory.add_messages([HumanMessage(content=input_str), AIMessage(content=output_str)])async def asave_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:"""异步保存对话上下文到记忆缓冲区"""input_str, output_str = self._get_input_output(inputs, outputs)await self.chat_memory.aadd_messages([HumanMessage(content=input_str), AIMessage(content=output_str)])def clear(self) -> None:"""清除记忆中的对话历史"""self.chat_memory.clear()async def aclear(self) -> None:"""异步清除记忆中的对话历史"""await self.chat_memory.aclear()