diff --git a/src/memos/memories/textual/tree_text_memory/organize/conflict.py b/src/memos/memories/textual/tree_text_memory/organize/conflict.py index 0ecf1847e..48620f73c 100644 --- a/src/memos/memories/textual/tree_text_memory/organize/conflict.py +++ b/src/memos/memories/textual/tree_text_memory/organize/conflict.py @@ -167,10 +167,12 @@ def _resolve_in_graph( if not self.graph_store.edge_exists(new_from, new_to, edge["type"], direction="ANY"): self.graph_store.add_edge(new_from, new_to, edge["type"]) - self.graph_store.delete_node(conflict_a.id) - self.graph_store.delete_node(conflict_b.id) + self.graph_store.update_node(conflict_a.id, {"status": "archived"}) + self.graph_store.update_node(conflict_b.id, {"status": "archived"}) + self.graph_store.add_edge(conflict_a.id, merged.id, type="MERGED_TO") + self.graph_store.add_edge(conflict_b.id, merged.id, type="MERGED_TO") logger.debug( - f"Remove {conflict_a.id} and {conflict_b.id}, and inherit their edges to {merged.id}." + f"Archive {conflict_a.id} and {conflict_b.id}, and inherit their edges to {merged.id}." ) def _merge_metadata( diff --git a/src/memos/memories/textual/tree_text_memory/organize/redundancy.py b/src/memos/memories/textual/tree_text_memory/organize/redundancy.py index 7b56a8107..7beae7347 100644 --- a/src/memos/memories/textual/tree_text_memory/organize/redundancy.py +++ b/src/memos/memories/textual/tree_text_memory/organize/redundancy.py @@ -30,7 +30,7 @@ def detect( self, memory: TextualMemoryItem, top_k: int = 5, scope: str | None = None ) -> list[tuple[TextualMemoryItem, TextualMemoryItem]]: """ - Detect redundancy by finding the most similar items in the graph database based on embedding, then use LLM to judge conflict. + Detect redundancy by finding the most similar items in the graph database based on embedding, then use LLM to judge redundancy. Args: memory: The memory item (should have an embedding attribute or field). top_k: Number of top similar nodes to retrieve. @@ -49,7 +49,7 @@ def detect( for info in embedding_candidates_info if info["score"] >= self.EMBEDDING_THRESHOLD and info["id"] != memory.id ] - # 3. Judge conflicts using LLM + # 3. Judge redundancys using LLM embedding_candidates = self.graph_store.get_nodes(embedding_candidates_ids) redundant_pairs = [] for embedding_candidate in embedding_candidates: @@ -57,7 +57,7 @@ def detect( prompt = [ { "role": "system", - "content": "You are a conflict detector for memory items.", + "content": "You are a redundancy detector for memory items.", }, { "role": "user", @@ -71,12 +71,12 @@ def detect( if "yes" in result.lower(): redundant_pairs.append([memory, embedding_candidate]) if len(redundant_pairs): - conflict_text = "\n".join( + redundant_text = "\n".join( f'"{pair[0].memory!s}" <==REDUNDANCY==> "{pair[1].memory!s}"' for pair in redundant_pairs ) logger.warning( - f"Detected {len(redundant_pairs)} redundancies for memory {memory.id}\n {conflict_text}" + f"Detected {len(redundant_pairs)} redundancies for memory {memory.id}\n {redundant_text}" ) return redundant_pairs @@ -84,12 +84,12 @@ def resolve_two_nodes(self, memory_a: TextualMemoryItem, memory_b: TextualMemory """ Resolve detected redundancies between two memory items using LLM fusion. Args: - memory_a: The first conflicting memory item. - memory_b: The second conflicting memory item. + memory_a: The first redundant memory item. + memory_b: The second redundant memory item. Returns: A fused TextualMemoryItem representing the resolved memory. """ - + return # waiting for implementation # ———————————— 1. LLM generate fused memory ———————————— metadata_for_resolve = ["key", "background", "confidence", "updated_at"] metadata_1 = memory_a.metadata.model_dump_json(include=metadata_for_resolve) @@ -115,18 +115,10 @@ def resolve_two_nodes(self, memory_a: TextualMemoryItem, memory_b: TextualMemory try: answer = re.search(r"(.*?)", response, re.DOTALL) answer = answer.group(1).strip() - # —————— 2.1 Can't resolve conflict, hard update by comparing timestamp ———— - if len(answer) <= 10 and "no" in answer.lower(): - logger.warning( - f"Conflict between {memory_a.id} and {memory_b.id} could not be resolved. " - ) - self._hard_update(memory_a, memory_b) - # —————— 2.2 Conflict resolved, update metadata and memory ———— - else: - fixed_metadata = self._merge_metadata(answer, memory_a.metadata, memory_b.metadata) - merged_memory = TextualMemoryItem(memory=answer, metadata=fixed_metadata) - logger.info(f"Resolved result: {merged_memory}") - self._resolve_in_graph(memory_a, memory_b, merged_memory) + fixed_metadata = self._merge_metadata(answer, memory_a.metadata, memory_b.metadata) + merged_memory = TextualMemoryItem(memory=answer, metadata=fixed_metadata) + logger.info(f"Resolved result: {merged_memory}") + self._resolve_in_graph(memory_a, memory_b, merged_memory) except json.decoder.JSONDecodeError: logger.error(f"Failed to parse LLM response: {response}") @@ -145,29 +137,14 @@ def resolve_one_node(self, memory: TextualMemoryItem) -> None: ) logger.debug(f"Merged memory: {memory.memory}") - def _hard_update(self, memory_a: TextualMemoryItem, memory_b: TextualMemoryItem): - """ - Hard update: compare updated_at, keep the newer one, overwrite the older one's metadata. - """ - time_a = datetime.fromisoformat(memory_a.metadata.updated_at) - time_b = datetime.fromisoformat(memory_b.metadata.updated_at) - - newer_mem = memory_a if time_a >= time_b else memory_b - older_mem = memory_b if time_a >= time_b else memory_a - - self.graph_store.delete_node(older_mem.id) - logger.warning( - f"Delete older memory {older_mem.id}: <{older_mem.memory}> due to conflict with {newer_mem.id}: <{newer_mem.memory}>" - ) - def _resolve_in_graph( self, - conflict_a: TextualMemoryItem, - conflict_b: TextualMemoryItem, + redundant_a: TextualMemoryItem, + redundant_b: TextualMemoryItem, merged: TextualMemoryItem, ): - edges_a = self.graph_store.get_edges(conflict_a.id, type="ANY", direction="ANY") - edges_b = self.graph_store.get_edges(conflict_b.id, type="ANY", direction="ANY") + edges_a = self.graph_store.get_edges(redundant_a.id, type="ANY", direction="ANY") + edges_b = self.graph_store.get_edges(redundant_b.id, type="ANY", direction="ANY") all_edges = edges_a + edges_b self.graph_store.add_node( @@ -175,18 +152,22 @@ def _resolve_in_graph( ) for edge in all_edges: - new_from = merged.id if edge["from"] in (conflict_a.id, conflict_b.id) else edge["from"] - new_to = merged.id if edge["to"] in (conflict_a.id, conflict_b.id) else edge["to"] + new_from = ( + merged.id if edge["from"] in (redundant_a.id, redundant_b.id) else edge["from"] + ) + new_to = merged.id if edge["to"] in (redundant_a.id, redundant_b.id) else edge["to"] if new_from == new_to: continue # Check if the edge already exists before adding if not self.graph_store.edge_exists(new_from, new_to, edge["type"], direction="ANY"): self.graph_store.add_edge(new_from, new_to, edge["type"]) - self.graph_store.delete_node(conflict_a.id) - self.graph_store.delete_node(conflict_b.id) + self.graph_store.update_node(redundant_a.id, {"status": "archived"}) + self.graph_store.update_node(redundant_b.id, {"status": "archived"}) + self.graph_store.add_edge(redundant_a.id, merged.id, type="MERGED_TO") + self.graph_store.add_edge(redundant_b.id, merged.id, type="MERGED_TO") logger.debug( - f"Remove {conflict_a.id} and {conflict_b.id}, and inherit their edges to {merged.id}." + f"Archive {redundant_a.id} and {redundant_b.id}, and inherit their edges to {merged.id}." ) def _merge_metadata( diff --git a/src/memos/memories/textual/tree_text_memory/organize/reorganizer.py b/src/memos/memories/textual/tree_text_memory/organize/reorganizer.py index 3bd81da49..16e1b3451 100644 --- a/src/memos/memories/textual/tree_text_memory/organize/reorganizer.py +++ b/src/memos/memories/textual/tree_text_memory/organize/reorganizer.py @@ -164,9 +164,9 @@ def handle_add(self, message: QueueMessage): logger.info(f"Resolved conflict between {added_node.id} and {existing_node.id}.") # ———————— 2. check for redundancy ———————— - redundancy = self.redundancy.detect(added_node, scope=added_node.metadata.memory_type) - if redundancy: - for added_node, existing_node in redundancy: + redundancies = self.redundancy.detect(added_node, scope=added_node.metadata.memory_type) + if redundancies: + for added_node, existing_node in redundancies: self.redundancy.resolve_two_nodes(added_node, existing_node) logger.info(f"Resolved redundancy between {added_node.id} and {existing_node.id}.") @@ -176,7 +176,7 @@ def handle_remove(self, message: QueueMessage): def handle_merge(self, message: QueueMessage): after_node = message.after_node[0] logger.debug(f"Handling merge operation: <{after_node.memory}>") - self.redundancy_resolver.resolve_one_node(after_node) + self.redundancy.resolve_one_node(after_node) def optimize_structure( self,