Method process_textual_memory

src/memos/mem_os/core.py:741–794

()

        logger.debug(f"Mem-reader mode is: {sync_mode}")

        def process_textual_memory():
            if (
                (messages is not None)
                and self.config.enable_textual_memory
                and self.mem_cubes[mem_cube_id].text_mem
            ):
                if self.mem_cubes[mem_cube_id].config.text_mem.backend != "tree_text":
                    add_memory = []
                    metadata = TextualMemoryMetadata(
                        user_id=target_user_id, session_id=target_session_id, source="conversation"
                    )
                    for message in messages:
                        add_memory.append(
                            TextualMemoryItem(memory=message["content"], metadata=metadata)
                        )
                    self.mem_cubes[mem_cube_id].text_mem.add(add_memory)
                else:
                    messages_list = [messages]
                    memories = self.mem_reader.get_memory(
                        messages_list,
                        type="chat",
                        info={"user_id": target_user_id, "session_id": target_session_id},
                        mode="fast" if sync_mode == "async" else "fine",
                    )
                    memories_flatten = [m for m_list in memories for m in m_list]
                    mem_ids: list[str] = self.mem_cubes[mem_cube_id].text_mem.add(memories_flatten)
                    logger.info(
                        f"Added memory user {target_user_id} to memcube {mem_cube_id}: {mem_ids}"
                    )
                    # submit messages for scheduler
                    if self.enable_mem_scheduler and self.mem_scheduler is not None:
                        if sync_mode == "async":
                            message_item = ScheduleMessageItem(
                                user_id=target_user_id,
                                mem_cube_id=mem_cube_id,
                                label=MEM_READ_TASK_LABEL,
                                content=json.dumps(mem_ids),
                                timestamp=datetime.utcnow(),
                                task_id=task_id,
                            )
                            self.mem_scheduler.submit_messages(messages=[message_item])
                        else:
                            message_item = ScheduleMessageItem(
                                user_id=target_user_id,
                                mem_cube_id=mem_cube_id,
                                label=ADD_TASK_LABEL,
                                content=json.dumps(mem_ids),
                                timestamp=datetime.utcnow(),
                                task_id=task_id,
                            )
                            logger.info(
                                f"[DIAGNOSTIC] core.add: Submitting message to scheduler: {message_item.model_dump_json(indent=2)}"
                            )
                            self.mem_scheduler.submit_messages(messages=[message_item])

        def process_preference_memory():
            if (