Source code for langchain.memory.combined

import warnings
from typing import Any, Dict, List, Set

from langchain_core.memory import BaseMemory
from langchain_core.pydantic_v1 import validator

from langchain.memory.chat_memory import BaseChatMemory


class CombinedMemory(BaseMemory):
    """将多个内存数据合并在一起。"""

    memories: List[BaseMemory]
    """用于跟踪所有应该被访问的记忆。"""

    @validator("memories")
    def check_repeated_memory_variable(
        cls, value: List[BaseMemory]
    ) -> List[BaseMemory]:
        all_variables: Set[str] = set()
        for val in value:
            overlap = all_variables.intersection(val.memory_variables)
            if overlap:
                raise ValueError(
                    f"The same variables {overlap} are found in multiple"
                    "memory object, which is not allowed by CombinedMemory."
                )
            all_variables |= set(val.memory_variables)

        return value

    @validator("memories")
    def check_input_key(cls, value: List[BaseMemory]) -> List[BaseMemory]:
        """检查记忆是否为BaseChatMemory类型，以确保输入键存在。"""
        for val in value:
            if isinstance(val, BaseChatMemory):
                if val.input_key is None:
                    warnings.warn(
                        "When using CombinedMemory, "
                        "input keys should be so the input is known. "
                        f" Was not set on {val}"
                    )
        return value

    @property
    def memory_variables(self) -> List[str]:
        """这个实例提供的所有内存变量。"""
        """Collected from the all the linked memories."""

        memory_variables = []

        for memory in self.memories:
            memory_variables.extend(memory.memory_variables)

        return memory_variables

    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, str]:
        """从子内存加载所有变量。"""
        memory_data: Dict[str, Any] = {}

        # Collect vars from all sub-memories
        for memory in self.memories:
            data = memory.load_memory_variables(inputs)
            for key, value in data.items():
                if key in memory_data:
                    raise ValueError(
                        f"The variable {key} is repeated in the CombinedMemory."
                    )
                memory_data[key] = value

        return memory_data

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
        """为每个内存保存此会话的上下文。"""
        # Save context for all sub-memories
        for memory in self.memories:
            memory.save_context(inputs, outputs)

    def clear(self) -> None:
        """清除此会话中的所有内存上下文。"""
        for memory in self.memories:
            memory.clear()