Langchain Postgres

LangChain Postgres Skill

Expert assistance for langchain-postgres: pgvector-backed vector store and PostgreSQL chat history for LangChain. Use v2 API (PGVectorStore + PGEngine) for new projects; v1 (PGVector) is legacy.

Install:

pip install -U langchain-postgres psycopg[binary] psycopg-pool
# Start PostgreSQL with pgvector:
docker run -p 5432:5432 -e POSTGRES_PASSWORD=password pgvector/pgvector:pg16

Reference: references/api.md (500 KB — full API reference).

When to Use This Skill

Activate when:

• Creating a pgvector store (v2) — using PGEngine + PGVectorStore.create() or create_sync()
• Using legacy PGVector (v1) — constructing PGVector with connection_string directly
• Initializing the vector table — calling engine.init_vectorstore_table() before first use
• Adding HNSW or IVFFlat indexes — using apply_vector_index() with HNSWIndex or IVFFlatIndex
• Hybrid search — configuring HybridSearchConfig with weighted_sum_ranking or reciprocal_rank_fusion
• Adding custom metadata columns — using Column / ColumnDict in table initialization
• Persisting chat history in Postgres — using PostgresChatMessageHistory with sync or async psycopg connections
• Async vector store operations — using AsyncPGVectorStore or the async methods on PGVectorStore
• Managing indexes — calling drop_vector_index(), reindex(), is_valid_index()

Quick Reference

v2 API — PGEngine + PGVectorStore (recommended for new projects)

from langchain_postgres.v2.engine import PGEngine
from langchain_postgres.v2.vectorstores import PGVectorStore
from langchain_openai import OpenAIEmbeddings

# Step 1: create engine (manages async connection pool)
engine = PGEngine.from_connection_string(
    url="postgresql+asyncpg://user:password@localhost:5432/mydb"
)

# Step 2: initialize the table (run once)
engine.init_vectorstore_table(
    table_name="my_vectors",
    vector_size=1536,           # must match embedding dimension
    # overwrite_existing=False  # set True to recreate
)

# Step 3: create the vector store (use factory, NOT constructor)
vector_store = PGVectorStore.create_sync(
    engine=engine,
    embedding_service=OpenAIEmbeddings(),
    table_name="my_vectors",
)

# Add documents
from langchain_core.documents import Document
docs = [Document(page_content="LangChain is a framework.", metadata={"source": "docs"})]
vector_store.add_documents(docs)

# Search
results = vector_store.similarity_search("What is LangChain?", k=3)

# Cleanup
engine.close()

v2 API — async usage

import asyncio
from langchain_postgres.v2.engine import PGEngine
from langchain_postgres.v2.vectorstores import PGVectorStore
from langchain_openai import OpenAIEmbeddings

async def main():
    engine = PGEngine.from_connection_string(
        url="postgresql+asyncpg://user:password@localhost:5432/mydb"
    )
    await engine.ainit_vectorstore_table("my_vectors", vector_size=1536)

    store = await PGVectorStore.create(
        engine=engine,
        embedding_service=OpenAIEmbeddings(),
        table_name="my_vectors",
    )

    await store.aadd_documents(docs)
    results = await store.asimilarity_search("LangChain", k=3)
    await engine.close()

asyncio.run(main())

v2 API — HNSW index for fast approximate search

from langchain_postgres.v2.indexes import HNSWIndex, HNSWQueryOptions, DistanceStrategy

# Apply HNSW index after loading data
vector_store.apply_vector_index(
    HNSWIndex(
        name="hnsw_idx",
        distance_strategy=DistanceStrategy.COSINE,
        m=16,           # max connections per node (default 16)
        ef_construction=64,  # build-time search width (default 64)
    )
)

# Use optimized query options at search time
results = vector_store.similarity_search(
    "my query",
    k=5,
    query_options=HNSWQueryOptions(ef_search=40),
)

# IVFFlat for larger datasets
from langchain_postgres.v2.indexes import IVFFlatIndex, IVFFlatQueryOptions
vector_store.apply_vector_index(
    IVFFlatIndex(name="ivfflat_idx", lists=100)
)
results = vector_store.similarity_search(
    "my query", k=5, query_options=IVFFlatQueryOptions(probes=10)
)

v2 API — hybrid search (vector + full-text)

from langchain_postgres.v2.hybrid_search_config import (
    HybridSearchConfig,
    reciprocal_rank_fusion,
    weighted_sum_ranking,
)

# Configure hybrid search on the store
vector_store.apply_hybrid_search_index(
    HybridSearchConfig(
        ranking_function=reciprocal_rank_fusion,   # or weighted_sum_ranking
        # For weighted_sum_ranking:
        # ranking_function=weighted_sum_ranking,
        # alpha=0.7,  # weight for vector score (0=full-text only, 1=vector only)
    )
)

results = vector_store.similarity_search("my query", k=5, search_type="hybrid")

v2 API — custom metadata columns

from langchain_postgres.v2.engine import PGEngine, Column

engine.init_vectorstore_table(
    table_name="my_vectors",
    vector_size=1536,
    metadata_columns=[
        Column(name="source", data_type="TEXT"),
        Column(name="created_at", data_type="TIMESTAMP"),
        Column(name="score", data_type="FLOAT"),
    ],
)

v1 API — legacy PGVector (simple setup)

from langchain_postgres import PGVector
from langchain_openai import OpenAIEmbeddings

# Direct constructor (v1 only)
vector_store = PGVector(
    embeddings=OpenAIEmbeddings(),
    collection_name="my_collection",
    connection="postgresql+psycopg://user:password@localhost:5432/mydb",
    # distance_strategy=DistanceStrategy.COSINE,
    # use_jsonb=True,
)

vector_store.add_documents(docs)
results = vector_store.similarity_search("LangChain", k=3)

PostgresChatMessageHistory (sync and async)

import psycopg
from langchain_postgres import PostgresChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

# Create table (run once)
conn_info = "postgresql://user:password@localhost:5432/mydb"
sync_conn = psycopg.connect(conn_info)
PostgresChatMessageHistory.create_tables(sync_conn, "chat_history")

# Use in a session
history = PostgresChatMessageHistory(
    table_name="chat_history",
    session_id="user-session-xyz",
    sync_connection=sync_conn,
)
history.add_messages([
    HumanMessage(content="Hello!"),
    AIMessage(content="Hi! How can I help?"),
])
print(history.messages)
history.clear()

# Async version
async def async_history():
    async_conn = await psycopg.AsyncConnection.connect(conn_info)
    await PostgresChatMessageHistory.acreate_tables(async_conn, "chat_history")
    history = PostgresChatMessageHistory(
        table_name="chat_history",
        session_id="user-abc",
        async_connection=async_conn,
    )
    await history.aadd_messages([HumanMessage(content="Hi")])
    msgs = await history.aget_messages()

v1 vs v2 Decision Guide

API Reference

PGEngine (v2)

PGVectorStore (v2) key methods

Reference Files

Requires: PostgreSQL with pgvector extension (pgvector/pgvector:pg16 Docker image).
Source: https://reference.langchain.com/python/langchain-postgres
GitHub: https://github.com/langchain-ai/langchain-postgres