feat: Support timestamp filtering for archival memories [LET-3469] (#4330)

Finish temporal filtering

Author: mattzh72

letta/functions/function_sets/base.py

@@ -78,16 +78,37 @@ async def archival_memory_insert(self: "Agent", content: str, tags: Optional[lis
 
 
 async def archival_memory_search(
-    self: "Agent", query: str, tags: Optional[list[str]] = None, tag_match_mode: Literal["any", "all"] = "any", top_k: Optional[int] = None
+    self: "Agent",
+    query: str,
+    tags: Optional[list[str]] = None,
+    tag_match_mode: Literal["any", "all"] = "any",
+    top_k: Optional[int] = None,
+    start_datetime: Optional[str] = None,
+    end_datetime: Optional[str] = None,
 ) -> Optional[str]:
     """
-    Search archival memory using semantic (embedding-based) search.
+    Search archival memory using semantic (embedding-based) search with optional temporal filtering.
 
     Args:
         query (str): String to search for using semantic similarity.
         tags (Optional[list[str]]): Optional list of tags to filter search results. Only passages with these tags will be returned.
         tag_match_mode (Literal["any", "all"]): How to match tags - "any" to match passages with any of the tags, "all" to match only passages with all tags. Defaults to "any".
         top_k (Optional[int]): Maximum number of results to return. Uses system default if not specified.
+        start_datetime (Optional[str]): Filter results to passages created after this datetime. ISO 8601 format: "YYYY-MM-DD" or "YYYY-MM-DDTHH:MM". Examples: "2024-01-15", "2024-01-15T14:30".
+        end_datetime (Optional[str]): Filter results to passages created before this datetime. ISO 8601 format: "YYYY-MM-DD" or "YYYY-MM-DDTHH:MM". Examples: "2024-01-20", "2024-01-20T17:00".
+
+    Examples:
+        # Search all passages
+        archival_memory_search(query="project updates")
+
+        # Search with date range (full days)
+        archival_memory_search(query="meetings", start_datetime="2024-01-15", end_datetime="2024-01-20")
+
+        # Search with specific time range
+        archival_memory_search(query="error logs", start_datetime="2024-01-15T09:30", end_datetime="2024-01-15T17:30")
+
+        # Search from a specific point in time onwards
+        archival_memory_search(query="customer feedback", start_datetime="2024-01-15T14:00")
 
     Returns:
         str: Query result string containing matching passages with timestamps and content.

letta/helpers/tpuf_client.py

@@ -167,6 +167,8 @@ async def query_passages(
         tag_match_mode: TagMatchMode = TagMatchMode.ANY,
         vector_weight: float = 0.5,
         fts_weight: float = 0.5,
+        start_date: Optional[datetime] = None,
+        end_date: Optional[datetime] = None,
     ) -> List[Tuple[PydanticPassage, float]]:
         """Query passages from Turbopuffer using vector search, full-text search, or hybrid search.
 
@@ -180,6 +182,8 @@ async def query_passages(
             tag_match_mode: TagMatchMode.ANY (match any tag) or TagMatchMode.ALL (match all tags) - default: TagMatchMode.ANY
             vector_weight: Weight for vector search results in hybrid mode (default: 0.5)
             fts_weight: Weight for FTS results in hybrid mode (default: 0.5)
+            start_date: Optional datetime to filter passages created after this date
+            end_date: Optional datetime to filter passages created before this date
 
         Returns:
             List of (passage, score) tuples
@@ -225,15 +229,38 @@ async def query_passages(
                         # For ANY mode, use ContainsAny to match any of the tags
                         tag_filter = ("tags", "ContainsAny", tags)
 
+                # build date filter conditions
+                date_filters = []
+                if start_date:
+                    # Turbopuffer expects datetime objects directly for comparison
+                    date_filters.append(("created_at", "Gte", start_date))
+                if end_date:
+                    # Turbopuffer expects datetime objects directly for comparison
+                    date_filters.append(("created_at", "Lte", end_date))
+
+                # combine all filters
+                all_filters = []
+                if tag_filter:
+                    all_filters.append(tag_filter)
+                if date_filters:
+                    all_filters.extend(date_filters)
+
+                # create final filter expression
+                final_filter = None
+                if len(all_filters) == 1:
+                    final_filter = all_filters[0]
+                elif len(all_filters) > 1:
+                    final_filter = ("And", all_filters)
+
                 if search_mode == "timestamp":
                     # Fallback: retrieve most recent passages by timestamp
                     query_params = {
                         "rank_by": ("created_at", "desc"),  # Order by created_at in descending order
                         "top_k": top_k,
                         "include_attributes": ["text", "organization_id", "archive_id", "created_at", "tags"],
                     }
-                    if tag_filter:
-                        query_params["filters"] = tag_filter
+                    if final_filter:
+                        query_params["filters"] = final_filter
 
                     result = await namespace.query(**query_params)
                     return self._process_single_query_results(result, archive_id, tags)
@@ -245,8 +272,8 @@ async def query_passages(
                         "top_k": top_k,
                         "include_attributes": ["text", "organization_id", "archive_id", "created_at", "tags"],
                     }
-                    if tag_filter:
-                        query_params["filters"] = tag_filter
+                    if final_filter:
+                        query_params["filters"] = final_filter
 
                     result = await namespace.query(**query_params)
                     return self._process_single_query_results(result, archive_id, tags)
@@ -258,8 +285,8 @@ async def query_passages(
                         "top_k": top_k,
                         "include_attributes": ["text", "organization_id", "archive_id", "created_at", "tags"],
                     }
-                    if tag_filter:
-                        query_params["filters"] = tag_filter
+                    if final_filter:
+                        query_params["filters"] = final_filter
 
                     result = await namespace.query(**query_params)
                     return self._process_single_query_results(result, archive_id, tags, is_fts=True)
@@ -274,8 +301,8 @@ async def query_passages(
                         "top_k": top_k,
                         "include_attributes": ["text", "organization_id", "archive_id", "created_at", "tags"],
                     }
-                    if tag_filter:
-                        vector_query["filters"] = tag_filter
+                    if final_filter:
+                        vector_query["filters"] = final_filter
                     queries.append(vector_query)
 
                     # full-text search query
@@ -284,8 +311,8 @@ async def query_passages(
                         "top_k": top_k,
                         "include_attributes": ["text", "organization_id", "archive_id", "created_at", "tags"],
                     }
-                    if tag_filter:
-                        fts_query["filters"] = tag_filter
+                    if final_filter:
+                        fts_query["filters"] = final_filter
                     queries.append(fts_query)
 
                     # execute multi-query

letta/services/agent_manager.py

@@ -2690,6 +2690,8 @@ async def query_agent_passages_async(
                         top_k=limit,
                         tags=tags,
                         tag_match_mode=tag_match_mode or TagMatchMode.ANY,
+                        start_date=start_date,
+                        end_date=end_date,
                     )
 
                     # Return just the passages (without scores)

letta/services/tool_executor/core_tool_executor.py

@@ -126,20 +126,70 @@ async def archival_memory_search(
         tags: Optional[list[str]] = None,
         tag_match_mode: Literal["any", "all"] = "any",
         top_k: Optional[int] = None,
+        start_datetime: Optional[str] = None,
+        end_datetime: Optional[str] = None,
     ) -> Optional[str]:
         """
-        Search archival memory using semantic (embedding-based) search.
+        Search archival memory using semantic (embedding-based) search with optional temporal filtering.
 
         Args:
             query (str): String to search for using semantic similarity.
             tags (Optional[list[str]]): Optional list of tags to filter search results. Only passages with these tags will be returned.
             tag_match_mode (Literal["any", "all"]): How to match tags - "any" to match passages with any of the tags, "all" to match only passages with all tags. Defaults to "any".
             top_k (Optional[int]): Maximum number of results to return. Uses system default if not specified.
+            start_datetime (Optional[str]): Filter results to passages created after this datetime. ISO 8601 format.
+            end_datetime (Optional[str]): Filter results to passages created before this datetime. ISO 8601 format.
 
         Returns:
             str: Query result string containing matching passages with timestamps, content, and tags.
         """
         try:
+            # Parse datetime parameters if provided
+            from datetime import datetime
+
+            start_date = None
+            end_date = None
+
+            if start_datetime:
+                try:
+                    # Try parsing as full datetime first (with time)
+                    start_date = datetime.fromisoformat(start_datetime)
+                except ValueError:
+                    try:
+                        # Fall back to date-only format
+                        start_date = datetime.strptime(start_datetime, "%Y-%m-%d")
+                        # Set to beginning of day
+                        start_date = start_date.replace(hour=0, minute=0, second=0, microsecond=0)
+                    except ValueError:
+                        raise ValueError(
+                            f"Invalid start_datetime format: {start_datetime}. Use ISO 8601 format (YYYY-MM-DD or YYYY-MM-DDTHH:MM)"
+                        )
+
+                # Apply agent's timezone if datetime is naive
+                if start_date.tzinfo is None and agent_state.timezone:
+                    tz = ZoneInfo(agent_state.timezone)
+                    start_date = start_date.replace(tzinfo=tz)
+
+            if end_datetime:
+                try:
+                    # Try parsing as full datetime first (with time)
+                    end_date = datetime.fromisoformat(end_datetime)
+                except ValueError:
+                    try:
+                        # Fall back to date-only format
+                        end_date = datetime.strptime(end_datetime, "%Y-%m-%d")
+                        # Set to end of day for end dates
+                        end_date = end_date.replace(hour=23, minute=59, second=59, microsecond=999999)
+                    except ValueError:
+                        raise ValueError(
+                            f"Invalid end_datetime format: {end_datetime}. Use ISO 8601 format (YYYY-MM-DD or YYYY-MM-DDTHH:MM)"
+                        )
+
+                # Apply agent's timezone if datetime is naive
+                if end_date.tzinfo is None and agent_state.timezone:
+                    tz = ZoneInfo(agent_state.timezone)
+                    end_date = end_date.replace(tzinfo=tz)
+
             # Convert string to TagMatchMode enum
             tag_mode = TagMatchMode.ANY if tag_match_mode == "any" else TagMatchMode.ALL
 
@@ -154,6 +204,8 @@ async def archival_memory_search(
                 embed_query=True,
                 tags=tags,
                 tag_match_mode=tag_mode,
+                start_date=start_date,
+                end_date=end_date,
             )
 
             # Format results to include tags with friendly timestamps