Merge pull request #41 from redis/fix/hash-deduplication-issue-31

Fix hash-based deduplication FT.AGGREGATE query execution

Author: abrookins

CLAUDE.md

@@ -1,5 +1,9 @@
 # CLAUDE.md - Redis Agent Memory Server Project Context
 
+## Redis Version
+This project uses Redis 8, which is the redis:8 docker image.
+Do not use Redis Stack or other earlier versions of Redis.
+
 ## Frequently Used Commands
 Get started in a new environment by installing `uv`:
 ```bash
@@ -188,7 +192,6 @@ EMBEDDING_MODEL=text-embedding-3-small
 
 # Memory Configuration
 LONG_TERM_MEMORY=true
-WINDOW_SIZE=20
 ENABLE_TOPIC_EXTRACTION=true
 ENABLE_NER=true
 ```

agent-memory-client/agent_memory_client/client.py

@@ -209,7 +209,6 @@ async def get_working_memory(
         session_id: str,
         user_id: str | None = None,
         namespace: str | None = None,
-        window_size: int | None = None,
         model_name: ModelNameLiteral | None = None,
         context_window_max: int | None = None,
     ) -> WorkingMemoryResponse:
@@ -220,7 +219,6 @@ async def get_working_memory(
             session_id: The session ID to retrieve working memory for
             user_id: The user ID to retrieve working memory for
             namespace: Optional namespace for the session
-            window_size: Optional number of messages to include
             model_name: Optional model name to determine context window size
             context_window_max: Optional direct specification of context window tokens
 
@@ -241,9 +239,6 @@ async def get_working_memory(
         elif self.config.default_namespace is not None:
             params["namespace"] = self.config.default_namespace
 
-        if window_size is not None:
-            params["window_size"] = str(window_size)
-
         # Use provided model_name or fall back to config default
         effective_model_name = model_name or self.config.default_model_name
         if effective_model_name is not None:
@@ -2139,7 +2134,6 @@ async def memory_prompt(
         query: str,
         session_id: str | None = None,
         namespace: str | None = None,
-        window_size: int | None = None,
         model_name: str | None = None,
         context_window_max: int | None = None,
         long_term_search: dict[str, Any] | None = None,
@@ -2154,7 +2148,6 @@ async def memory_prompt(
             query: The input text to find relevant context for
             session_id: Optional session ID to include session messages
             namespace: Optional namespace for the session
-            window_size: Optional number of messages to include
             model_name: Optional model name to determine context window size
             context_window_max: Optional direct specification of context window tokens
             long_term_search: Optional search parameters for long-term memory
@@ -2169,7 +2162,6 @@ async def memory_prompt(
             prompt = await client.memory_prompt(
                 query="What are my UI preferences?",
                 session_id="current_session",
-                window_size=10,
                 long_term_search={
                     "topics": {"any": ["preferences", "ui"]},
                     "limit": 5
@@ -2190,8 +2182,6 @@ async def memory_prompt(
                 session_params["namespace"] = namespace
             elif self.config.default_namespace is not None:
                 session_params["namespace"] = self.config.default_namespace
-            if window_size is not None:
-                session_params["window_size"] = str(window_size)
             # Use provided model_name or fall back to config default
             effective_model_name = model_name or self.config.default_model_name
             if effective_model_name is not None:

agent_memory_server/api.py

@@ -582,19 +582,16 @@ async def memory_prompt(
     logger.debug(f"Memory prompt params: {params}")
 
     if params.session:
-        # Use token limit for memory prompt, fallback to message count for backward compatibility
+        # Use token limit for memory prompt - model info is required now
         if params.session.model_name or params.session.context_window_max:
             token_limit = _get_effective_token_limit(
                 model_name=params.session.model_name,
                 context_window_max=params.session.context_window_max,
             )
-            effective_window_size = (
-                token_limit  # We'll handle token-based truncation below
-            )
+            effective_token_limit = token_limit
         else:
-            effective_window_size = (
-                params.session.window_size
-            )  # Fallback to message count
+            # No model info provided - use all messages without truncation
+            effective_token_limit = None
         working_mem = await working_memory.get_working_memory(
             session_id=params.session.session_id,
             namespace=params.session.namespace,
@@ -616,46 +613,42 @@ async def memory_prompt(
                     )
                 )
             # Apply token-based truncation if model info is provided
-            if params.session.model_name or params.session.context_window_max:
+            if effective_token_limit is not None:
                 # Token-based truncation
                 if (
                     _calculate_messages_token_count(working_mem.messages)
-                    > effective_window_size
+                    > effective_token_limit
                 ):
                     # Keep removing oldest messages until we're under the limit
                     recent_messages = working_mem.messages[:]
                     while len(recent_messages) > 1:  # Always keep at least 1 message
                         recent_messages = recent_messages[1:]  # Remove oldest
                         if (
                             _calculate_messages_token_count(recent_messages)
-                            <= effective_window_size
+                            <= effective_token_limit
                         ):
                             break
                 else:
                     recent_messages = working_mem.messages
-
-                for msg in recent_messages:
-                    if msg.role == "user":
-                        msg_class = base.UserMessage
-                    else:
-                        msg_class = base.AssistantMessage
-                    _messages.append(
-                        msg_class(
-                            content=TextContent(type="text", text=msg.content),
-                        )
-                    )
             else:
-                # No token-based truncation - use all messages
-                for msg in working_mem.messages:
-                    if msg.role == "user":
-                        msg_class = base.UserMessage
-                    else:
-                        msg_class = base.AssistantMessage
-                    _messages.append(
-                        msg_class(
-                            content=TextContent(type="text", text=msg.content),
-                        )
+                # No token limit provided - use all messages
+                recent_messages = working_mem.messages
+
+            for msg in recent_messages:
+                if msg.role == "user":
+                    msg_class = base.UserMessage
+                elif msg.role == "assistant":
+                    msg_class = base.AssistantMessage
+                else:
+                    # For tool messages or other roles, treat as assistant for MCP compatibility
+                    # since MCP base only supports UserMessage and AssistantMessage
+                    msg_class = base.AssistantMessage
+
+                _messages.append(
+                    msg_class(
+                        content=TextContent(type="text", text=msg.content),
                     )
+                )
 
     if params.long_term_search:
         logger.debug(

agent_memory_server/config.py

@@ -52,7 +52,7 @@ class Settings(BaseSettings):
     long_term_memory: bool = True
     openai_api_key: str | None = None
     anthropic_api_key: str | None = None
-    generation_model: str = "gpt-4o-mini"
+    generation_model: str = "gpt-4o"
     embedding_model: str = "text-embedding-3-small"
     port: int = 8000
     mcp_port: int = 9000
@@ -118,7 +118,6 @@ class Settings(BaseSettings):
     auth0_client_secret: str | None = None
 
     # Working memory settings
-    window_size: int = 20  # Default number of recent messages to return
     summarization_threshold: float = (
         0.7  # Fraction of context window that triggers summarization
     )

agent_memory_server/long_term_memory.py

@@ -324,21 +324,37 @@ async def compact_long_term_memories(
             index_name = Keys.search_index_name()
 
             # Create aggregation query to group by memory_hash and find duplicates
-            agg_query = (
-                f"FT.AGGREGATE {index_name} {filter_str} "
-                "GROUPBY 1 @memory_hash "
-                "REDUCE COUNT 0 AS count "
-                'FILTER "@count>1" '  # Only groups with more than 1 memory
-                "SORTBY 2 @count DESC "
-                f"LIMIT 0 {limit}"
-            )
+            agg_query = [
+                "FT.AGGREGATE",
+                index_name,
+                filter_str,
+                "GROUPBY",
+                str(1),
+                "@memory_hash",
+                "REDUCE",
+                "COUNT",
+                str(0),
+                "AS",
+                "count",
+                "FILTER",
+                "@count>1",  # Only groups with more than 1 memory
+                "SORTBY",
+                str(2),
+                "@count",
+                "DESC",
+                "LIMIT",
+                str(0),
+                str(limit),
+            ]
 
             # Execute aggregation to find duplicate groups
-            duplicate_groups = await redis_client.execute_command(agg_query)
+            duplicate_groups = await redis_client.execute_command(*agg_query)
 
             if duplicate_groups and duplicate_groups[0] > 0:
                 num_groups = duplicate_groups[0]
-                logger.info(f"Found {num_groups} groups of hash-based duplicates")
+                logger.info(
+                    f"Found {num_groups} groups with hash-based duplicates to process"
+                )
 
                 # Process each group of duplicates
                 for i in range(1, len(duplicate_groups), 2):
@@ -423,9 +439,11 @@ async def compact_long_term_memories(
                                 )
                     except Exception as e:
                         logger.error(f"Error processing duplicate group: {e}")
+            else:
+                logger.info("No hash-based duplicates found")
 
             logger.info(
-                f"Completed hash-based deduplication. Merged {memories_merged} memories."
+                f"Completed hash-based deduplication. Removed {memories_merged} duplicate memories."
             )
         except Exception as e:
             logger.error(f"Error during hash-based duplicate compaction: {e}")

agent_memory_server/main.py

@@ -135,7 +135,6 @@ async def lifespan(app: FastAPI):
 
     logger.info(
         "Redis Agent Memory Server initialized",
-        window_size=settings.window_size,
         generation_model=settings.generation_model,
         embedding_model=settings.embedding_model,
         long_term_memory=settings.long_term_memory,

agent_memory_server/mcp.py

@@ -467,7 +467,6 @@ async def memory_prompt(
     query: str,
     session_id: SessionId | None = None,
     namespace: Namespace | None = None,
-    window_size: int = settings.window_size,
     model_name: ModelNameLiteral | None = None,
     context_window_max: int | None = None,
     topics: Topics | None = None,
@@ -579,7 +578,6 @@ async def memory_prompt(
             session_id=_session_id,
             namespace=namespace.eq if namespace and namespace.eq else None,
             user_id=user_id.eq if user_id and user_id.eq else None,
-            window_size=window_size,
             model_name=model_name,
             context_window_max=context_window_max,
         )

agent_memory_server/models.py

@@ -7,7 +7,6 @@
 from pydantic import BaseModel, Field
 from ulid import ULID
 
-from agent_memory_server.config import settings
 from agent_memory_server.filters import (
     CreatedAt,
     Entities,
@@ -238,7 +237,6 @@ class WorkingMemoryRequest(BaseModel):
     session_id: str
     namespace: str | None = None
     user_id: str | None = None
-    window_size: int = settings.window_size
     model_name: ModelNameLiteral | None = None
     context_window_max: int | None = None