Merge pull request #12 from FlowLLM-AI/vector_store_update

feat(vector): add list filter support for metadata queries

Author: jinliyl

flowllm/core/enumeration/__init__.py

@@ -1,14 +1,12 @@
 """Core enumeration module."""
 
 from .chunk_enum import ChunkEnum
-from .content_block_type import ContentBlockType
 from .http_enum import HttpEnum
 from .registry_enum import RegistryEnum
 from .role import Role
 
 __all__ = [
     "ChunkEnum",
-    "ContentBlockType",
     "HttpEnum",
     "RegistryEnum",
     "Role",

flowllm/core/enumeration/content_block_type.py

@@ -6,7 +6,7 @@
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from .tool_call import ToolCall
-from ..enumeration import Role, ContentBlockType
+from ..enumeration import Role
 
 
 class ContentBlock(BaseModel):
@@ -35,7 +35,7 @@ class ContentBlock(BaseModel):
 
     model_config = ConfigDict(extra="allow")
 
-    type: ContentBlockType = Field(default="")
+    type: str = Field(default="")
     content: str | dict | list = Field(default="")
 
     @model_validator(mode="before")
@@ -51,8 +51,8 @@ def init_block(cls, data: dict):
     def simple_dump(self) -> dict:
         """Convert ContentBlock to a simple dictionary format."""
         result = {
-            "type": self.type.value,
-            self.type.value: self.content,
+            "type": self.type,
+            self.type: self.content,
             **self.model_extra,
         }
 
@@ -95,11 +95,6 @@ def simple_dump(self, add_reasoning: bool = True) -> dict:
 
         return result
 
-    @property
-    def string_buffer(self) -> str:
-        """Get a string representation of the message with role and content."""
-        return f"{self.role.value}: {self.dump_content()}"
-
 
 class Trajectory(BaseModel):
     """Represents a conversation trajectory with messages and optional scoring."""

flowllm/core/schema/message.py

@@ -87,6 +87,7 @@ def _build_chroma_filters(
                 - Term filters: {"key": "value"} -> exact match
                 - Range filters: {"key": {"gte": 1, "lte": 10}} -> range query
                 - unique_id: Filters by document ID (stored separately from metadata)
+                - Keys can use "metadata." prefix (will be stripped for ChromaDB)
 
         Returns:
             Tuple of (where_clause, ids_filter):
@@ -108,6 +109,12 @@ def _build_chroma_filters(
                     ids_filter = [filter_value]
                 continue
 
+            # Strip "metadata." prefix if present (ChromaDB stores metadata fields directly)
+            if key.startswith("metadata."):
+                chroma_key = key[len("metadata.") :]
+            else:
+                chroma_key = key
+
             if isinstance(filter_value, dict):
                 # Range filter: {"gte": 1, "lte": 10}
                 range_conditions = {}
@@ -120,10 +127,13 @@ def _build_chroma_filters(
                 if "lt" in filter_value:
                     range_conditions["$lt"] = filter_value["lt"]
                 if range_conditions:
-                    where_conditions[key] = range_conditions
+                    where_conditions[chroma_key] = range_conditions
+            elif isinstance(filter_value, list):
+                # List filter: use $in operator for OR logic
+                where_conditions[chroma_key] = {"$in": filter_value}
             else:
                 # Term filter: direct value comparison
-                where_conditions[key] = filter_value
+                where_conditions[chroma_key] = filter_value
 
         return (where_conditions if where_conditions else None, ids_filter)
 

flowllm/core/vector_store/chroma_vector_store.py

@@ -158,6 +158,9 @@ def _build_es_filters(filter_dict: Optional[Dict[str, Any]] = None) -> List[Dict
                     range_conditions["lt"] = filter_value["lt"]
                 if range_conditions:
                     filters.append({"range": {es_key: range_conditions}})
+            elif isinstance(filter_value, list):
+                # List filter: use terms query for OR logic
+                filters.append({"terms": {es_key: filter_value}})
             else:
                 # Term filter: direct value comparison
                 filters.append({"term": {es_key: filter_value}})

flowllm/core/vector_store/es_vector_store.py

@@ -228,6 +228,10 @@ def _matches_filters(node: VectorNode, filter_dict: dict = None) -> bool:
                     range_match = False
                 if not range_match:
                     return False
+            elif isinstance(filter_value, list):
+                # List filter: value must match any item in the list (OR logic)
+                if value not in filter_value:
+                    return False
             else:
                 # Term filter: direct value comparison
                 if value != filter_value:

flowllm/core/vector_store/memory_vector_store.py

@@ -218,6 +218,16 @@ def _build_sql_filters(
                     param_idx += 1
                 if range_conditions:
                     conditions.append(f"({' AND '.join(range_conditions)})")
+            elif isinstance(filter_value, list):
+                # List filter: use IN clause for OR logic
+                if use_async:
+                    placeholders = ", ".join(f"${param_idx + i}" for i in range(len(filter_value)))
+                    conditions.append(f"{jsonb_path} IN ({placeholders})")
+                else:
+                    placeholders = ", ".join(["%s"] * len(filter_value))
+                    conditions.append(f"{jsonb_path} IN ({placeholders})")
+                params.extend([str(v) for v in filter_value])
+                param_idx += len(filter_value)
             else:
                 # Term filter: direct value comparison
                 if use_async:

flowllm/core/vector_store/pgvector_vector_store.py

@@ -181,7 +181,7 @@ def _build_qdrant_filters(filter_dict: Optional[Dict[str, Any]] = None):
             filter_dict = {"age": {"gte": 18, "lte": 65}}
             ```
         """
-        from qdrant_client.http.models import FieldCondition, MatchValue, Range
+        from qdrant_client.http.models import FieldCondition, MatchAny, MatchValue, Range
 
         if not filter_dict:
             return None
@@ -215,6 +215,14 @@ def _build_qdrant_filters(filter_dict: Optional[Dict[str, Any]] = None):
                             range=Range(**range_conditions),
                         ),
                     )
+            elif isinstance(filter_value, list):
+                # List filter: use MatchAny for OR logic
+                conditions.append(
+                    FieldCondition(
+                        key=qdrant_key,
+                        match=MatchAny(any=filter_value),
+                    ),
+                )
             else:
                 # Term filter: direct value comparison
                 conditions.append(

flowllm/core/vector_store/qdrant_vector_store.py

@@ -167,14 +167,23 @@ def create_sample_nodes(workspace_id: str, prefix: str = "") -> List[VectorNode]
             VectorNode(
                 unique_id=f"{id_prefix}node3",
                 workspace_id=workspace_id,
+                content="Machine learning is a subset of artificial intelligence.",
+                metadata={
+                    "node_type": "tech_new",
+                    "category": "ML",
+                },
+            ),
+            VectorNode(
+                unique_id=f"{id_prefix}node4",
+                workspace_id=workspace_id,
                 content="I love eating delicious seafood, especially fresh fish.",
                 metadata={
                     "node_type": "food",
                     "category": "preference",
                 },
             ),
             VectorNode(
-                unique_id=f"{id_prefix}node4",
+                unique_id=f"{id_prefix}node5",
                 workspace_id=workspace_id,
                 content="Deep learning uses neural networks with multiple layers.",
                 metadata={
@@ -277,17 +286,20 @@ def test_search(self, workspace_id: str):
     def test_search_with_filter(self, workspace_id: str):
         """Test vector search with filter."""
         logger.info("=" * 20 + " FILTER SEARCH TEST " + "=" * 20)
-        filter_dict = {"metadata.node_type": "tech"}
+        filter_dict = {"metadata.node_type": ["tech", "tech_new"]}
         results = self.client.search(
             "What is artificial intelligence?",
             workspace_id=workspace_id,
             top_k=5,
             filter_dict=filter_dict,
         )
-        logger.info(f"Filtered search returned {len(results)} results (node_type=tech)")
+        logger.info(f"Filtered search returned {len(results)} results (node_type in [tech, tech_new])")
         for i, r in enumerate(results, 1):
             logger.info(f"Filtered Result {i}: {r.model_dump(exclude={'vector'})}")
-            assert r.metadata.get("node_type") == "tech", "All results should have node_type=tech"
+            assert r.metadata.get("node_type") in [
+                "tech",
+                "tech_new",
+            ], "All results should have node_type in [tech, tech_new]"
 
     def test_search_with_id(self, workspace_id: str):
         """Test vector search by unique_id with empty query."""
@@ -503,17 +515,20 @@ async def test_search(self, workspace_id: str):
     async def test_search_with_filter(self, workspace_id: str):
         """Test async vector search with filter."""
         logger.info("ASYNC - " + "=" * 20 + " FILTER SEARCH TEST " + "=" * 20)
-        filter_dict = {"metadata.node_type": "tech"}
+        filter_dict = {"metadata.node_type": ["tech", "tech_new"]}
         results = await self.client.async_search(
             "What is artificial intelligence?",
             workspace_id=workspace_id,
             top_k=5,
             filter_dict=filter_dict,
         )
-        logger.info(f"Filtered search returned {len(results)} results (node_type=tech)")
+        logger.info(f"Filtered search returned {len(results)} results (node_type in [tech, tech_new])")
         for i, r in enumerate(results, 1):
             logger.info(f"Filtered Result {i}: {r.model_dump(exclude={'vector'})}")
-            assert r.metadata.get("node_type") == "tech", "All results should have node_type=tech"
+            assert r.metadata.get("node_type") in [
+                "tech",
+                "tech_new",
+            ], "All results should have node_type in [tech, tech_new]"
 
     async def test_search_with_id(self, workspace_id: str):
         """Test async vector search by unique_id with empty query."""