diff --git a/.env.template b/.env.template index d4c1fa4e..736bf749 100644 --- a/.env.template +++ b/.env.template @@ -232,13 +232,6 @@ LLM_ANTHROPIC_API_KEY=your-anthropic-api-key-here # ============================================================================= # Vector Store Settings # ============================================================================= -# VECTOR_STORE_TYPE="lancedb" -# VECTOR_STORE_NAMESPACE="honcho" -# VECTOR_STORE_TURBOPUFFER_API_KEY= -# VECTOR_STORE_TURBOPUFFER_REGION= -# VECTOR_STORE_LANCEDB_PATH="./lancedb_data" - -# Vector store settings # Vector store type: "pgvector", "turbopuffer", or "lancedb" VECTOR_STORE_TYPE=pgvector @@ -249,14 +242,14 @@ VECTOR_STORE_MIGRATED=false # Namespaces follow the pattern: # - Documents: {NAMESPACE}.{workspace}.{observer}.{observed} # - Messages: {NAMESPACE}.{workspace}.messages -VECTOR_STORE_NAMESPACE=honcho +# VECTOR_STORE_NAMESPACE=honcho # Inherits from NAMESPACE if not set # Embedding dimensions (default: 1536 for OpenAI text-embedding-3-small) -VECTOR_STORE_DIMENSIONS=1536 +# VECTOR_STORE_DIMENSIONS=1536 # Turbopuffer-specific settings (required if TYPE is "turbopuffer") # VECTOR_STORE_TURBOPUFFER_API_KEY=your-turbopuffer-api-key -# VECTOR_STORE_TURBOPUFFER_REGION=us-east-1 +# VECTOR_STORE_TURBOPUFFER_REGION=gcp-us-east4 # LanceDB-specific settings (local embedded mode) -VECTOR_STORE_LANCEDB_PATH=./lancedb_data +# VECTOR_STORE_LANCEDB_PATH=./lancedb_data diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index 6ffa7b63..87ae5831 100644 --- a/src/deriver/queue_manager.py +++ b/src/deriver/queue_manager.py @@ -51,7 +51,7 @@ class WorkerOwnership(NamedTuple): QUEUE_CLEANUP_INTERVAL_SECONDS = 43200 # 12 hours -RECONCILIATION_INTERVAL_SECONDS = 900 # 15 minutes +RECONCILIATION_INTERVAL_SECONDS = 300 # 5 minutes class QueueManager: diff --git a/src/deriver/vector_reconciliation.py b/src/deriver/vector_reconciliation.py index 8b3dcddf..fb7cec8c 100644 --- a/src/deriver/vector_reconciliation.py +++ b/src/deriver/vector_reconciliation.py @@ -479,7 +479,6 @@ async def run_vector_reconciliation_cycle() -> ReconciliationMetrics: from src.crud.document import cleanup_soft_deleted_documents - print("Running vector reconciliation cycle") async with tracked_db("reconciliation") as db: # If no external vector store (pgvector mode), only clean up soft-deleted documents if external_vector_store is None: diff --git a/src/vector_store/lancedb.py b/src/vector_store/lancedb.py index fe4d3a58..5176b2f6 100644 --- a/src/vector_store/lancedb.py +++ b/src/vector_store/lancedb.py @@ -6,6 +6,7 @@ This module provides a LanceDB-based implementation of the VectorStore interface import asyncio import logging +from collections.abc import Sequence from typing import Any, cast import lancedb @@ -250,6 +251,10 @@ class LanceDBVectorStore(VectorStore): """ Convert a filter dict to SQL WHERE clause syntax. + Supports filter formats: + - {"key": "value"} -> key = 'value' + - {"key": {"in": ["a", "b"]}} -> key IN ('a', 'b') + Args: filters: Dictionary of attribute -> value filters @@ -261,8 +266,20 @@ class LanceDBVectorStore(VectorStore): conditions: list[str] = [] for key, value in filters.items(): + # Check if value is a dict with "in" operator + if isinstance(value, dict) and "in" in value: + # IN clause for list membership + in_values = cast(Sequence[Any], value["in"]) + if in_values: + escaped_values = [ + f"'{str(v).replace(chr(39), chr(39) + chr(39))}'" + if isinstance(v, str) + else str(v) + for v in in_values + ] + conditions.append(f"{key} IN ({', '.join(escaped_values)})") # Handle string values with proper quoting - if isinstance(value, str): + elif isinstance(value, str): # Escape single quotes in the value escaped_value = value.replace("'", "''") conditions.append(f"{key} = '{escaped_value}'") diff --git a/src/vector_store/turbopuffer.py b/src/vector_store/turbopuffer.py index ba5c3612..fe209d77 100644 --- a/src/vector_store/turbopuffer.py +++ b/src/vector_store/turbopuffer.py @@ -6,7 +6,7 @@ This module provides a Turbopuffer-based implementation of the VectorStore inter import logging from collections.abc import Sequence -from typing import Any, Literal +from typing import Any, Literal, cast from turbopuffer import AsyncTurbopuffer, NotFoundError from turbopuffer.lib.namespace import AsyncNamespace @@ -18,8 +18,9 @@ from . import VectorQueryResult, VectorRecord, VectorStore, VectorUpsertResult logger = logging.getLogger(__name__) -# Type alias for Turbopuffer's equality filter format +# Type aliases for Turbopuffer's filter formats EqFilter = tuple[str, Literal["Eq"], Any] +InFilter = tuple[str, Literal["In"], Sequence[Any]] AndFilter = tuple[Literal["And"], Sequence[Filter]] DISTANCE_METRIC = "cosine_distance" @@ -192,8 +193,13 @@ class TurbopufferVectorStore(VectorStore): Convert a filter dict to Turbopuffer filter format. Turbopuffer uses tuples like (attribute, "Eq", value) for filters, + (attribute, "In", [values]) for membership filters, and ("And", [filters]) for combining multiple filters. + Supports filter formats: + - {"key": "value"} -> ("key", "Eq", "value") + - {"key": {"in": ["a", "b"]}} -> ("key", "In", ["a", "b"]) + Args: filters: Dictionary of attribute -> value filters @@ -203,11 +209,16 @@ class TurbopufferVectorStore(VectorStore): if not filters: return None - filter_list: list[EqFilter] = [] + filter_list: list[EqFilter | InFilter] = [] for key, value in filters.items(): - # Simple equality filter using "Eq" operator - eq_filter: EqFilter = (key, "Eq", value) - filter_list.append(eq_filter) + # Check if value is a dict with "in" operator + if isinstance(value, dict) and "in" in value: + # Membership filter using "In" operator + in_values = cast(Sequence[Any], value["in"]) + filter_list.append((key, "In", in_values)) + else: + # Simple equality filter using "Eq" operator + filter_list.append((key, "Eq", cast(Any, value))) if not filter_list: return None diff --git a/tests/integration/test_message_embeddings.py b/tests/integration/test_message_embeddings.py index de544ee0..691d12c1 100644 --- a/tests/integration/test_message_embeddings.py +++ b/tests/integration/test_message_embeddings.py @@ -12,12 +12,26 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from src import models +from src.config import settings from src.crud import create_messages from src.models import Peer, Workspace from src.schemas import MessageCreate from src.utils.search import search +def _stores_embeddings_in_postgres() -> bool: + """Check if current config stores MessageEmbedding rows in postgres.""" + return settings.VECTOR_STORE.TYPE == "pgvector" or not settings.VECTOR_STORE.MIGRATED + + +# Skip tests that depend on MessageEmbedding rows when using external store in migrated mode +requires_postgres_embeddings = pytest.mark.skipif( + not _stores_embeddings_in_postgres(), + reason="MessageEmbedding rows not created when TYPE != 'pgvector' and MIGRATED=true", +) + + +@requires_postgres_embeddings @pytest.mark.asyncio async def test_message_embedding_created_when_setting_enabled( db_session: AsyncSession, @@ -123,6 +137,7 @@ async def test_message_embedding_not_created_when_setting_disabled( assert embedding_record is None +@requires_postgres_embeddings @pytest.mark.asyncio async def test_multiple_message_embeddings_created_when_setting_enabled( db_session: AsyncSession, @@ -182,6 +197,7 @@ async def test_multiple_message_embeddings_created_when_setting_enabled( assert embedding_record.peer_name == test_peer.name +@requires_postgres_embeddings @pytest.mark.asyncio async def test_semantic_search_when_embeddings_enabled( db_session: AsyncSession, @@ -257,6 +273,7 @@ async def test_semantic_search_when_embeddings_enabled( assert created_message.public_id in found_message_ids +@requires_postgres_embeddings @pytest.mark.asyncio async def test_message_chunking_creates_multiple_embeddings( db_session: AsyncSession, diff --git a/tests/utils/test_agent_tools.py b/tests/utils/test_agent_tools.py index da3c9a89..b54098e0 100644 --- a/tests/utils/test_agent_tools.py +++ b/tests/utils/test_agent_tools.py @@ -312,9 +312,14 @@ class TestSearchMemory: """Tests for _handle_search_memory.""" async def test_returns_matching_observations( - self, make_tool_context: Callable[..., ToolContext] + self, + make_tool_context: Callable[..., ToolContext], + monkeypatch: pytest.MonkeyPatch, ): """Returns observations matching semantic query.""" + # Force pgvector queries since test documents are created directly in postgres + monkeypatch.setattr("src.config.settings.VECTOR_STORE.MIGRATED", False) + ctx = make_tool_context() result = await _handle_search_memory(ctx, {"query": "coffee preferences"})