Method memorize

methods/memos_agent.py:231–351

Store memory using MemOS's official extract() and add() methods. This uses MemOS's native LLM-based memory extraction pipeline. Args: text: Input text to memorize. **kwargs: Additional arguments (unused). Returns: MemoryBuildRes

(self, text: str, **kwargs)

        return system

    def memorize(self, text: str, **kwargs) -> MemoryBuildResult:
        """Store memory using MemOS's official extract() and add() methods.

        This uses MemOS's native LLM-based memory extraction pipeline.

        Args:
            text: Input text to memorize.
            **kwargs: Additional arguments (unused).

        Returns:
            MemoryBuildResult with extraction and storage details.
        """
        context_id = self._get_context_id()
        memory_system = self._get_memory_system(context_id)

        # Truncate input text
        bounded_text = self._truncate_to_tokens(text, self.max_input_tokens)

        # Convert text to message format for MemOS extract()
        messages = [{"role": "user", "content": bounded_text}]

        start_time = time.time()
        all_extraction_results: List[Dict[str, Any]] = []
        all_stored_memories: List[Dict[str, Any]] = []
        extraction_time = 0.0
        add_time = 0.0

        try:
            # Use MemOS's official extract() method
            # This calls MemOS's LLM-based extractor internally
            extracted_items = memory_system.extract(messages)
            extraction_time = time.time() - start_time

            # Record extraction results
            for item in extracted_items:
                memory_text = getattr(item, "memory", str(item))
                metadata = getattr(item, "metadata", {})
                if hasattr(metadata, "model_dump"):
                    metadata = metadata.model_dump()

                all_extraction_results.append({
                    "memory": memory_text,
                    "metadata": metadata,
                })

            # Use MemOS's official add() method to store extracted memories
            if extracted_items:
                add_start = time.time()
                memory_system.add(extracted_items)
                add_time = time.time() - add_start

                for item in extracted_items:
                    memory_text = getattr(item, "memory", str(item))
                    metadata = getattr(item, "metadata", {})
                    if hasattr(metadata, "model_dump"):
                        metadata = metadata.model_dump()

                    all_stored_memories.append({