Plan: Vector Search (As-Built)

Related spec: ./spec.md Last synced with code: 2026-04-10

1. Hexagonal Mapping

Layer	Component	File
Domain Port	`IVectorSearch`	`domain/ports/search/vector_search.py`
Infrastructure Adapter	`PgVectorSearchAdapter`	`infrastructure/adapters/search/pgvector_search_adapter.py`

2. Port

class IVectorSearch(ABC):
    async def search_datasets(
        self,
        query_embedding: list[float],
        limit: int = 10,
        portal_filter: str | None = None,
        min_similarity: float = 0.55,
    ) -> list[SearchResult]: ...

    async def search_datasets_hybrid(
        self,
        query_embedding: list[float],
        query_text: str,
        limit: int = 10,
        portal_filter: str | None = None,
        rrf_k: int = 60,
    ) -> list[SearchResult]: ...

    async def index_dataset(
        self,
        dataset_id: str,
        content: str,
        embedding: list[float],
    ) -> None: ...

    async def delete_dataset_chunks(self, dataset_id: str) -> None: ...

3. SearchResult

@dataclass
class SearchResult:
    dataset_id: str
    title: str
    description: str
    portal: str
    download_url: str
    columns: dict  # JSON
    score: float   # 0.0 - 1.0

4. Vector Search SQL (pseudocode)

WITH query_embedding AS (
    SELECT CAST(:emb AS vector) AS vec
),
ranked_chunks AS (
    SELECT
        dc.dataset_id,
        dc.content,
        1 - (dc.embedding <=> (SELECT vec FROM query_embedding)) AS similarity,
        ROW_NUMBER() OVER (
            PARTITION BY dc.dataset_id
            ORDER BY dc.embedding <=> (SELECT vec FROM query_embedding)
        ) AS rank
    FROM dataset_chunks dc
    JOIN datasets d ON dc.dataset_id = d.id
    WHERE (:portal_filter IS NULL OR d.portal = :portal_filter)
)
SELECT *
FROM ranked_chunks rc
JOIN datasets d ON rc.dataset_id = d.id
WHERE rc.rank = 1  -- best chunk per dataset
  AND rc.similarity >= :min_similarity
ORDER BY rc.similarity DESC
LIMIT :limit

5. Hybrid Search (BM25 + Vector RRF)

async def search_datasets_hybrid(query_embedding, query_text, limit, ...):
    # 1. BM25 ranking
    bm25_results = await self._bm25_search(query_text, limit * 2)
    # 2. Vector ranking
    vector_results = await self.search_datasets(query_embedding, limit * 2)
    # 3. RRF fusion
    fused = {}
    for rank, r in enumerate(bm25_results):
        fused[r.dataset_id] = fused.get(r.dataset_id, 0) + 1 / (rrf_k + rank + 1)
    for rank, r in enumerate(vector_results):
        fused[r.dataset_id] = fused.get(r.dataset_id, 0) + 1 / (rrf_k + rank + 1)
    # 4. Sort by combined score, return top-limit
    return sorted(fused.items(), key=lambda x: -x[1])[:limit]

6. Persistence

`dataset_chunks`

id UUID PK
dataset_id FK → datasets(id)
content TEXT
embedding vector(1024)
created_at TIMESTAMPTZ

Indexes:

HNSW: dataset_chunks(embedding vector_cosine_ops)
BTREE: dataset_chunks(dataset_id)
GIN (FTS, for BM25): over content — TBD whether it exists

7. Embedding Strategy (from collector_tasks)

Per dataset, 3 chunks with distinct content:

main: {title} — {description}
columns: {columns_joined}
contextual: {sample_rows_text} (row sample)

Each chunk has its own embedding generated by Cohere Embed v3 with input_type="search_document".

8. Source Files

domain/ports/search/vector_search.py
infrastructure/adapters/search/pgvector_search_adapter.py
Used by: application/pipeline/connectors/vector_search.py

9. Deviations from Constitution

Principle VII (Observability): no search quality metrics.

End of plan.md

FilesExpand file tree

plan.md

Latest commit

History