PostgreSQL + pgvector

PgVectorStore provides persistent, collection-scoped RAG storage using PostgreSQL and pgvector. It supports exact or HNSW vector retrieval, JSONB metadata filters, PostgreSQL full-text search, hybrid rank fusion, and custom schema mappings without changing normal RAG calls.

Use it when an application already operates PostgreSQL or needs database-level transactions, roles, backups, monitoring, connection pools, and Row-Level Security. PyAIStack manages vector-store behavior; the application remains responsible for operating PostgreSQL safely.

Requirements

Install the optional PostgreSQL dependency:

pip install "pyaistack[postgres]"

Enable pgvector in the database that PyAIStack will connect to:

CREATE EXTENSION IF NOT EXISTS vector;

Extensions are database-specific. Confirm the selected database before troubleshooting a missing-extension error:

SELECT current_database(), current_user, extname, extversion
FROM pg_extension
WHERE extname = 'vector';

Managed schema quick start

initialize=True explicitly creates the mapped schema, tables, constraints, and indexes when they are missing. It does not install the pgvector extension and it does not drop or rewrite existing objects.

import os

from pyaistack import RAG
from pyaistack.vectorstores import PgVectorStore

with PgVectorStore(
    dsn=os.environ["PYAISTACK_POSTGRES_DSN"],
    dimensions=768,
    collection="support",
    initialize=True,
) as store:
    rag = RAG(vector_store=store)
    rag.add(
        ["Support is available Monday through Friday."],
        metadatas=[{"category": "support"}],
    )

Initialization is idempotent but deliberately strict. If an object already exists with incompatible columns, constraints, vector dimensions, or index requirements, construction raises VectorStoreSchemaError instead of silently modifying it.

Reopen and ask

Use the same collection, dimension, schema mapping, and embedding model. Normal query processes should omit initialize=True after deployment or migration.

import os

from pyaistack import RAG
from pyaistack.vectorstores import PgVectorStore

with PgVectorStore(
    dsn=os.environ["PYAISTACK_POSTGRES_DSN"],
    dimensions=768,
    collection="support",
) as store:
    rag = RAG(vector_store=store)
    answer = rag.ask("When is support available?")
    print(answer.text)

Inspect retrieval with search()

search() embeds the query and returns SearchResult objects without calling the chat model.

results = rag.search("support availability", top_k=5)

for result in results:
    print(result.score)
    print(result.ranking_score)
    print(result.document.metadata)
    print(result.document.text)

result.score is cosine similarity. In hybrid mode, result.ranking_score contains the separate reciprocal-rank-fusion score. search() returns a tuple and does not have an answer-level .text attribute.

Direct DSN connections

With dsn=..., PyAIStack opens, commits or rolls back, and closes one connection per operation:

import os

from pyaistack.vectorstores import PgVectorStore

store = PgVectorStore(
    dsn=os.environ["PYAISTACK_POSTGRES_DSN"],
    dimensions=768,
    collection="support",
)

The DSN can use a PostgreSQL URI or libpq keyword format. Configure TLS, connection routing, and credentials in that value according to application policy.

Application-owned connection factory

Use connection_factory when the application owns connection acquisition or an existing pool. Supply either dsn or connection_factory, never both.

from collections.abc import Iterator
from contextlib import contextmanager
import os

import psycopg

from pyaistack.vectorstores import PgVectorStore


@contextmanager
def connection_factory() -> Iterator[psycopg.Connection]:
    connection = psycopg.connect(os.environ["PYAISTACK_POSTGRES_DSN"])
    try:
        yield connection
    finally:
        connection.close()


store = PgVectorStore(
    connection_factory=connection_factory,
    dimensions=768,
    collection="support",
)

A factory may return a bare psycopg connection or a context manager such as a pool checkout. PyAIStack commits or rolls back each operation. A bare factory-provided connection remains application-owned; a context manager is exited after the operation. The application owns the pool lifecycle.

Exact and approximate vector search

PostgreSQL search mode is separate from RAG retrieval mode:

Approximate search is the default. The mapped table must have a valid HNSW or IVFFlat cosine index. A managed schema creates an HNSW cosine index by default.

import os

from pyaistack.vectorstores import PgVectorStore

store = PgVectorStore(
    dsn=os.environ["PYAISTACK_POSTGRES_DSN"],
    dimensions=768,
    collection="support",
    search_mode="approximate",
    hnsw_m=16,
    hnsw_ef_construction=64,
    hnsw_ef_search=40,
)

Use exact mode when deterministic exhaustive scoring matters more than latency, or when a compatible custom table has no ANN index:

import os

from pyaistack.vectorstores import PgVectorStore

store = PgVectorStore(
    dsn=os.environ["PYAISTACK_POSTGRES_DSN"],
    dimensions=768,
    collection="support",
    search_mode="exact",
    create_hnsw_index=False,
)

Managed HNSW indexes support at most 2,000 vector dimensions. Use exact mode or an application-designed compatible storage strategy for larger dimensions. Always measure recall and latency using representative application queries.

Hybrid retrieval

PostgreSQL hybrid mode combines:

1. cosine retrieval through pgvector;
2. plain-text PostgreSQL full-text retrieval using the simple text-search configuration; and
3. reciprocal-rank fusion using the same public ranking behavior as the SQLite store.

from pyaistack import RAG, RAGConfig

rag = RAG(
    vector_store=store,
    config=RAGConfig(
        retrieval_mode="hybrid",
        top_k=5,
        candidate_k=20,
    ),
)

Both branches apply the collection and metadata predicates before their result limits. Candidate counts are bounded by the store's configured max_candidates setting. min_score remains a cosine-similarity threshold, not a fusion-score threshold.

Metadata filters

Metadata is stored as JSONB. Filters preserve the same scalar/list behavior as the in-memory and SQLite stores:

results = rag.search(
    "purchase approval",
    metadata_filter={
        "department": "finance",
        "tags": ["purchasing", "approval"],
    },
)

Every key must match. Comparisons are exact and case-sensitive. Missing fields do not equal JSON null. Filters are parameterized and applied in PostgreSQL before result limiting.

Managed schemas include a general JSONB GIN index. For frequent filters on a small set of known fields, inspect plans with EXPLAIN (ANALYZE, BUFFERS) and add application-owned expression indexes when evidence supports them.

Custom schema mapping

Map PyAIStack semantic fields to compatible application-owned names with PgVectorSchema:

import os

from pyaistack.vectorstores import PgVectorSchema, PgVectorStore

schema = PgVectorSchema(
    schema_name="application_ai",
    table_name="knowledge_chunks",
    collection_table_name="knowledge_collections",
    collection_column="namespace",
    id_column="chunk_id",
    text_column="content",
    metadata_column="attributes",
    vector_column="embedding",
    created_at_column="created_at",
    updated_at_column="updated_at",
    collection_created_at_column="created_at",
    model_column="embedding_model",
    dimensions_column="embedding_dimensions",
    schema_version_column="adapter_schema_version",
)

store = PgVectorStore(
    dsn=os.environ["PYAISTACK_POSTGRES_DSN"],
    dimensions=768,
    collection="support",
    schema=schema,
    initialize=False,
)

Set any timestamp mapping to None when the compatible table intentionally does not contain that optional timestamp. Identifiers must be simple PostgreSQL identifiers containing letters, numbers, and underscores; arbitrary SQL is not accepted.

Required document-table contract

The collection table requires a unique collection field plus model, dimension, and schema-version fields. Dimension and schema version must be non-null. The model field may initially be null and is bound by the first RAG configuration.

Approximate mode additionally requires a valid HNSW or IVFFlat index using vector_cosine_ops. Custom schema validation is read-only, but opening a store registers its selected collection in the compatible collection table.

Run the custom-schema example

Install PostgreSQL support, provide the database password through the environment, and pull the embedding model:

pip install -e ".[postgres]"
export PGPASSWORD="your-postgresql-password"
ollama pull embeddinggemma

The example uses the local connection defaults in examples/postgres/postgres_config.py: localhost:5432, user admin_user, and database test_pyaistack. The target database must already have the pgvector extension enabled.

Run it from the repository root:

python examples/postgres/custom_schema.py

On its first run, the example:

1. creates the application_ai schema;
2. creates the custom-named knowledge_collections and knowledge_chunks tables;
3. indexes two documents in the custom_schema_demo collection; and
4. runs a metadata-filtered rag.search() and prints each score, source, and matching text.

Later runs reuse the indexed documents. The example uses initialize=True so it is self-contained. For an existing production schema, apply reviewed migrations first and use initialize=False as shown above.

Model and dimension binding

dimensions must match the embedding model output and the mapped vector(n) column. PyAIStack durably binds each collection to one embedding model and dimension. Equal dimensions alone do not make two models compatible.

Opening a collection with another model or dimension fails before retrieval or insertion. Use a different collection and, for another dimension, a different table or schema. clear() removes only the selected collection's documents and preserves its model/dimension binding.

Transactions, limits, and cleanup

• Each add() call validates its inputs and commits atomically.
• Inserts use bounded batches inside the same transaction.
• Duplicate collection/document IDs fail instead of silently overwriting data.
• clear() is collection-scoped and never truncates the shared table.
• statement_timeout_ms bounds database operations.
• batch_size, max_candidates, and max_query_chars bound database work.
• close() is idempotent and prevents later use; a context manager is recommended.
• Write failures are rolled back and are not blindly retried.

Production checklist

Before deployment:

• install and pin a supported PostgreSQL and pgvector version;
• run schema initialization or reviewed migrations separately from normal query startup;
• use TLS and least-privilege database roles;
• configure a bounded application connection pool;
• set realistic connection and statement timeouts;
• enforce tenant access with application authorization and PostgreSQL RLS;
• monitor connection use, query latency, locks, table/index growth, dead rows, and replication health;
• run ANALYZE and normal PostgreSQL vacuum maintenance;
• benchmark exact versus ANN recall and tune HNSW from measured results;
• back up the database and regularly test restoration; and
• rehearse migrations and rollback using production-like data.

PyAIStack does not provision PostgreSQL, install extensions, manage passwords, operate replication, or replace database monitoring and disaster-recovery procedures.

Errors

Driver exceptions remain chained for debugging, while normal messages avoid including the DSN, document text, metadata values, vectors, or SQL parameters.

Runnable examples

Index first, then run the query or retrieval-only example:

python examples/postgres/index.py
python examples/postgres/query.py "What does policy FIN-042 require?"
python examples/postgres/search.py "FIN-042 budget policy" --top-k 5
python examples/postgres/connection_factory.py "How are backups handled?"
python examples/postgres/custom_schema.py

See examples/postgres/README.md for local setup and environment variables.