Temporality

docker-compose.yaml

@@ -169,4 +169,4 @@ services:
 
 networks:
   default:
-    name: openrag_default
+    name: openrag_default

docs/content/docs/documentation/API.mdx

@@ -78,6 +78,22 @@ Upload a new file to a specific partition for indexing.
 - `201 Created`: Returns task status URL
 - `409 Conflict`: File already exists in partition
 
+##### Temporal Filtering
+OpenRAG supports temporal filtering to retrieve documents from specific time periods.
+The client can include temporal fields to allow temporal aware search in search endpoints.
+
+4 temporal fields are automatically added to the schema of the collection:
+
+* `datetime`: ISO 8601 format date of the primary timestamp of when a file is created in your system
+* `created_at`: ISO 8601 format date of when the file was created
+* `updated_at`: ISO 8601 format date of when the file was last modified
+* `indexed_at`: ISO 8601 format date of when the file was indexed in the vector database
+
+:::info
+`datetime`, `created_at` and `updated_at` are provided by the client in the metadata of the file during upload, while `indexed_at` is automatically set by openRAG at indexing time.
+:::
+
+
 ##### Upload files while modeling relations between them
 
 OpenRAG supports document relationships to enable context-aware retrieval.
@@ -202,6 +218,9 @@ Perform semantic search across specified partitions.
 | `include_related` (optional)   | boolean | `false`   | Include chunks from files with same `relationship_id` |
 | `include_ancestors` (optional) | boolean | `false`   | Include chunks from ancestor files (via `parent_id` chain) |
 | `related_limit` (optional)     | integer | 20      | Max related/ancestor chunks to fetch per result (used when `include_related` or `include_ancestors` is true) |
+| `filter` (optional)            | string  | None    | Milvus filter expression string for additional filtering (optional). Supports comparison, range, and logical operators. |
+| `filter_params` (optional)     | object  | None    | Dictionary of parameter values for templated filters. Use with placeholders in filter expression for better performance. |
+
 
 **Responses:**
 - `200 OK`: JSON list of document links (HATEOAS format)
@@ -223,6 +242,8 @@ Search within a specific partition only.
 | `include_related` (optional)   | boolean | `false`   | Include chunks from files with same `relationship_id` |
 | `include_ancestors` (optional) | boolean | `false`   | Include chunks from ancestor files (via `parent_id` chain) |
 | `related_limit` (optional)     | integer | 20      | Max related/ancestor chunks to fetch per result (used when `include_related` or `include_ancestors` is true) |
+| `filter` (optional)            | string  | None    | Milvus filter expression string for additional filtering (optional). Supports comparison, range, and logical operators. |
+| `filter_params` (optional)     | object  | None    | Dictionary of parameter values for templated filters. Use with placeholders in filter expression for better performance. |
 
 **Response:** Same as multi-partition search
 
@@ -233,7 +254,7 @@ GET /search/partition/{partition}/file/{file_id}
 
 Search within a particular file in a partition.
 
-**Query Parameters:** Same as partition search
+**Query Parameters:** Same as partition search, including `filter` and `filter_params`.
 **Response:** Same as other search endpoints
 
 ---

docs/content/docs/documentation/milvus_migration.md

@@ -0,0 +1,128 @@
+---
+title: Milvus Migrations
+---
+
+# Milvus Version Migration
+OpenRAG has been upgraded from Milvus **2.5.4** to **2.6.11** to leverage the enhancements introduced in the latest releases, particularly the new temporal querying capabilities added in version **2.6.6+**.
+
+## What's New in 2.6.x
+
+Milvus 2.6.6+ introduced the **`TIMESTAMPTZ`** field type, which enables:
+
+- **Comparison and range filtering** using standard operators (`=`, `!=`, `<`, `>`, etc.)
+- **Interval arithmetic** — add or subtract durations (days, hours, minutes) directly in filter expressions
+- **Time-based indexing** for faster temporal queries
+- **Combined filtering** — pair timestamp conditions with vector similarity search
+
+**Example — basic comparison:**
+```python
+expr = "tsz != ISO '2025-01-03T00:00:00+08:00'"
+results = client.query(
+    collection_name, 
+    filter=expr,
+    output_fields=["id", "tsz"],
+    limit=10
+)
+```
+
+**Example — interval arithmetic:**
+```python
+expr = "tsz + INTERVAL 'P1D' > ISO '2025-01-03T00:00:00+08:00'"
+results = client.query(
+    collection_name,
+    filter=expr,
+    output_fields=["id", "tsz"], 
+    limit=10
+)
+```
+
+> `INTERVAL` values follow [ISO 8601 duration](https://en.wikipedia.org/wiki/ISO_8601#Durations) syntax: 
+> * `P1D` = 1 day
+> * `PT3H` = 3 hours
+> * `P2DT6H` = 2 days and 6 hours.
+
+## Current State
+
+:::info
+Temporal fields are currently stored as **strings**, not **`TIMESTAMPTZ`**. Migrating to `TIMESTAMPTZ` requires a schema and index change, and Milvus doesn't support migrations on schema and index changes: it has to be handled manually.
+
+Until a migration strategy is define, filtering still works via **lexicographic string comparison** on ISO 8601 strings:
+```python
+expr = "tsz != '2025-01-03T00:00:00+08:00'"  # No ISO/INTERVAL keywords
+results = client.query(
+    collection_name,
+    filter=expr,
+    output_fields=["id", "tsz"],
+    limit=10
+)
+```
+Full `TIMESTAMPTZ` support will be activated in a future release once the migration is established.
+:::
+
+## Milvus Update Steps
+These steps must be performed on a deployment running OpenRAG **prior to version 1.1.6** (Milvus 2.5.4).
+
+> For the full official reference, see the [Milvus upgrade guide](https://milvus.io/docs/upgrade_milvus_standalone-docker.md#Upgrade-process).
+
+### Step 1 — Upgrade to Milvus 2.5.16 first
+
+Milvus requires an intermediate upgrade to **v2.5.16** before jumping to 2.6.x. 
+
+Edit `vdb/milvus.yaml` and set the Milvus image tag:
+
+```diff lang=yaml
+// vdb/milvus.yaml
+milvus:
+-  image: milvusdb/milvus:v2.5.4
++  image: milvusdb/milvus:v2.5.16 # Migrate to milvus 2.5.16
+```
+
+Then restart the stack:
+
+```bash
+docker compose down
+docker compose up -d
+```
+
+Wait for all services to be healthy before continuing.
+
+### Step 2 — Upgrade to Milvus 2.6.11
+
+Update `vdb/milvus.yaml` with the target versions (MinIO must also be updated for compatibility):
+
+```diff lang=yaml
+// vdb/milvus.yaml
+minio:
+-  image: minio/minio:RELEASE.2023-03-20T20-16-18Z
++  image: minio/minio:RELEASE.2024-12-18T13-15-44Z
+
+...
+milvus:
+-  image: milvusdb/milvus:v2.5.16
++  image: milvusdb/milvus:v2.6.11
+```
+
+### Step 3 — Stop all services
+
+```bash
+docker compose down
+```
+
+Verify that all containers are stopped before proceeding:
+
+```bash
+docker ps | grep milvus
+```
+
+### Step 4 — Start with the new image
+
+```bash
+docker compose up -d
+```
+
+Once healthy, confirm the running version:
+
+```bash
+docker inspect milvus-standalone --format '{{ .Config.Image }}'
+# Expected: milvusdb/milvus:v2.6.11
+```

openrag/components/indexer/indexer.py

@@ -225,19 +225,20 @@ async def asearch(
         self,
         query: str,
         top_k: int = 5,
-        similarity_threshold: float = 0.80,
+        similarity_threshold: float = 0.60,
         partition: str | list[str] | None = None,
-        filter: dict | None = None,
+        filter: str | None = None,
+        filter_params: dict | None = None,
     ) -> list[Document]:
         partition_list = self._check_partition_list(partition)
-        filter = filter or {}
         vectordb = ray.get_actor("Vectordb", namespace="openrag")
         return await vectordb.async_search.remote(
             query=query,
             partition=partition_list,
             top_k=top_k,
             similarity_threshold=similarity_threshold,
             filter=filter,
+            filter_params=filter_params,
         )
 
     def _check_partition_str(self, partition: str | None) -> str:

openrag/components/indexer/vectordb/vectordb.py

@@ -1,6 +1,7 @@
 import asyncio
 import time
 from abc import ABC, abstractmethod
+from datetime import UTC, datetime
 
 import numpy as np
 import ray
@@ -139,6 +140,7 @@ def __init__(self):
 
             self.config = load_config()
             self.logger = get_logger()
+            self.time_fields = ["datetime", "created_at", "updated_at", "indexed_at"]
 
             # init milvus clients
             self.port = self.config.vectordb.get("port")
@@ -274,6 +276,9 @@ def _create_schema(self):
             dim=self.embedder.embedding_dimension,
         )
 
+        # for time_field in self.time_fields:
+        #     schema.add_field(field_name=time_field, datatype=DataType.TIMESTAMPTZ, nullable=True)
+
         if self.hybrid_search:
             # Add sparse field for BM25 - this will be auto-generated
             schema.add_field(
@@ -327,6 +332,13 @@ def _create_index(self):
                 "bm25_b": 0.75,
             },
         )
+        # # indexes for dates TIMESTAMPTZ field
+        # for time_field in self.time_fields:
+        #     index_params.add_index(
+        #         field_name=time_field,
+        #         index_type="STL_SORT",  # Index for TIMESTAMPTZ
+        #         index_name=f"{time_field}_idx",
+        #     )
 
         return index_params
 
@@ -370,11 +382,14 @@ async def async_add_documents(self, chunks: list[Document], user: dict) -> None:
             entities = []
             vectors = await self.embedder.aembed_documents(chunks)
             order_metadata_l: list[dict] = _gen_chunk_order_metadata(n=len(chunks))
+            indexed_at = datetime.now(UTC).isoformat()
+
             for chunk, vector, order_metadata in zip(chunks, vectors, order_metadata_l):
                 entities.append(
                     {
                         "text": chunk.page_content,
                         "vector": vector,
+                        "indexed_at": indexed_at,
                         **order_metadata,
                         **chunk.metadata,
                     }
@@ -386,6 +401,7 @@ async def async_add_documents(self, chunks: list[Document], user: dict) -> None:
             )
 
             # insert file_id and partition into partition_file_manager
+            file_metadata.update({"indexed_at": indexed_at})
             self.partition_file_manager.add_file_to_partition(
                 file_id=file_id,
                 partition=partition,
@@ -443,21 +459,22 @@ async def async_search(
         self,
         query: str,
         top_k: int = 5,
-        similarity_threshold: int = 0.80,
+        similarity_threshold: float = 0.80,
         partition: list[str] = None,
-        filter: dict | None = None,
+        filter: str | None = None,
+        filter_params: dict | None = None,
         with_surrounding_chunks: bool = False,
     ) -> list[Document]:
         expr_parts = []
         if partition != ["all"]:
             expr_parts.append(f"partition in {partition}")
 
         if filter:
-            for key, value in filter.items():
-                expr_parts.append(f"{key} == '{value}'")
+            expr_parts.append(filter)
 
         # Join all parts with " and " only if there are multiple conditions
         expr = " and ".join(expr_parts) if expr_parts else ""
+        filter_params = filter_params or {}
 
         try:
             query_vector = await self.embedder.aembed_query(query)
@@ -474,6 +491,7 @@ async def async_search(
                 },
                 "limit": top_k,
                 "expr": expr,
+                "expr_params": filter_params,
             }
             if self.hybrid_search:
                 sparse_param = {
@@ -485,6 +503,7 @@ async def async_search(
                     },
                     "limit": top_k,
                     "expr": expr,
+                    "expr_params": filter_params,
                 }
                 reqs = [
                     AnnSearchRequest(**vector_param),
@@ -498,10 +517,24 @@ async def async_search(
                     limit=top_k,
                 )
             else:
+                vector_param = {
+                    "data": [query_vector],
+                    "anns_field": "vector",
+                    "search_params": {
+                        "metric_type": "COSINE",
+                        "params": {
+                            "ef": 64,
+                            "radius": similarity_threshold,
+                            "range_filter": 1.0,
+                        },
+                    },
+                    "limit": top_k,
+                }
                 response = await self._async_client.search(
                     collection_name=self.collection_name,
                     output_fields=["*"],
-                    limit=top_k,
+                    filter=expr,
+                    filter_params=filter_params,
                     **vector_param,
                 )
 
@@ -612,29 +645,22 @@ async def get_file_chunks(self, file_id: str, partition: str, include_id: bool =
         log = self.logger.bind(file_id=file_id, partition=partition)
         try:
             self._check_file_exists(file_id, partition)
-            # Adjust filter expression based on the type of value
-            filter_expression = "partition == {partition} and file_id == {file_id}"
-            filter_params = {"partition": partition, "file_id": file_id}
-
-            # Pagination parameters
-            offset = 0
-            results = []
+            filter_expr = f'partition == "{partition}" and file_id == "{file_id}"'
             excluded_keys = ["text", "vector", "_id"] if not include_id else ["text", "vector"]
 
+            results = []
+            iterator = self._client.query_iterator(
+                collection_name=self.collection_name,
+                filter=filter_expr,
+                batch_size=limit,
+                output_fields=["*"],
+            )
             while True:
-                response = await self._async_client.query(
-                    collection_name=self.collection_name,
-                    filter=filter_expression,
-                    filter_params=filter_params,
-                    limit=limit,
-                    offset=offset,
-                )
-
-                if not response:
-                    break  # No more results
-
-                results.extend(response)
-                offset += len(response)  # Move offset forward
+                batch = iterator.next()
+                if not batch:
+                    iterator.close()
+                    break
+                results.extend(batch)
 
             docs = [
                 Document(