symfonic.memory.backends.postgres_vector

postgres_vector

PostgreSQL vector backend using pgvector.

Implements VectorBackend protocol with cosine distance (<=> operator). All operations enforce tenant isolation via WHERE tenant_id = $N.

asyncpg and pgvector must be installed.

pip install symfonic-core[postgres]

PostgresVectorBackend

PostgresVectorBackend(pool: PostgresPoolManager, embedding_dim: int = 1536, *, search_strategy: Literal['exact', 'approximate'] = 'exact')

VectorBackend backed by PostgreSQL + pgvector.

Uses cosine distance (<=> operator) for similarity search. All operations enforce tenant isolation via WHERE tenant_id = $N. search_strategy='exact' ranks the authorized subset: bounded returned rows, but O(visible vectors) database distance work. 'approximate' opts into HNSW, which can underfill or miss neighbours after scope filtering.

Source code in src/symfonic/memory/backends/postgres_vector.py

def __init__(self, pool: PostgresPoolManager, embedding_dim: int = 1536, *,
             search_strategy: Literal["exact", "approximate"] = "exact") -> None:
    _require_asyncpg("PostgresVectorBackend")
    if embedding_dim <= 0:
        raise ValueError("embedding_dim must be a positive integer")
    if search_strategy not in ("exact", "approximate"):
        raise ValueError("search_strategy must be exact or approximate")
    self._pool = pool
    self._embedding_dim = embedding_dim
    self.search_strategy = search_strategy

pool property

pool: PostgresPoolManager

The pool this backend speaks through, and so its transaction domain.

count async

count(scope: TenantScope) -> int

Return the total number of stored vectors for the tenant.

Source code in src/symfonic/memory/backends/postgres_vector.py

async def count(self, scope: TenantScope) -> int:
    """Return the total number of stored vectors for the tenant."""
    conn = await self._pool.acquire()
    try:
        row = await conn.fetchrow(
            "SELECT COUNT(*)::int AS n FROM memory_vectors WHERE tenant_id=$1",
            scope.tenant_id,
        )
        return int(row["n"]) if row else 0
    finally:
        await self._pool.release(conn)

ensure_schema async

ensure_schema() -> None

Create the memory_vectors table and indexes if not present.

Source code in src/symfonic/memory/backends/postgres_vector.py

async def ensure_schema(self) -> None:
    """Create the memory_vectors table and indexes if not present."""
    conn = await self._pool.acquire()
    try:
        await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        await conn.execute(VECTORS_DDL)
        await conn.execute(VECTORS_SCOPE_PATH_MIGRATION)
        for idx in VECTOR_INDEXES_DDL:
            await conn.execute(idx)
        await conn.execute(_ann_index_ddl(self._embedding_dim))
    finally:
        await self._pool.release(conn)