Class PreferenceTextMemory

src/memos/memories/textual/preference.py:33–344

Preference textual memory implementation for storing and retrieving memories.

class PreferenceTextMemory(BaseTextMemory):
    """Preference textual memory implementation for storing and retrieving memories."""

    def __init__(self, config: PreferenceTextMemoryConfig):
        """Initialize memory with the given configuration."""
        self.config: PreferenceTextMemoryConfig = config
        self.extractor_llm: OpenAILLM | OllamaLLM | AzureLLM = LLMFactory.from_config(
            config.extractor_llm
        )
        self.vector_db: MilvusVecDB | QdrantVecDB = VecDBFactory.from_config(config.vector_db)
        self.embedder: OllamaEmbedder | ArkEmbedder | SenTranEmbedder | UniversalAPIEmbedder = (
            EmbedderFactory.from_config(config.embedder)
        )
        self.reranker = RerankerFactory.from_config(config.reranker)

        self.extractor = ExtractorFactory.from_config(
            config.extractor,
            llm_provider=self.extractor_llm,
            embedder=self.embedder,
            vector_db=self.vector_db,
        )

        self.adder = AdderFactory.from_config(
            config.adder,
            llm_provider=self.extractor_llm,
            embedder=self.embedder,
            vector_db=self.vector_db,
        )
        self.retriever = RetrieverFactory.from_config(
            config.retriever,
            llm_provider=self.extractor_llm,
            embedder=self.embedder,
            reranker=self.reranker,
            vector_db=self.vector_db,
        )

    def get_memory(
        self, messages: list[MessageList], type: str, info: dict[str, Any], **kwargs
    ) -> list[TextualMemoryItem]:
        """Get memory based on the messages.
        Args:
            messages (list[MessageList]): The messages to get memory from.
            type (str): The type of memory to get.
            info (dict[str, Any]): The info to get memory.
            **kwargs: Additional keyword arguments to pass to the extractor.
        """
        return self.extractor.extract(messages, type, info, **kwargs)

    def search(
        self, query: str, top_k: int, info=None, search_filter=None, **kwargs
    ) -> list[TextualMemoryItem]:
        """Search for memories based on a query.
        Args:
            query (str): The query to search for.
            top_k (int): The number of top results to return.
            info (dict): Leave a record of memory consumption.
        Returns:
            list[TextualMemoryItem]: List of matching memories.