langchain-postgres

PostgresChatMessageHistory allows you to persist chat messages in a PostgreSQL table. It requires a table_name and a session_id (a unique identifier for the chat session, often a UUID).

Steps:

1. Establish a connection using psycopg (or psycopg.AsyncConnection for async).
2. Create the table schema once using PostgresChatMessageHistory.create_tables(connection, table_name).
3. Initialize PostgresChatMessageHistory with the connection, table name, and session ID.
4. Use .add_messages() to save messages and .messages to retrieve them.

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

conn_info = "..."
sync_connection = psycopg.connect(conn_info)

table_name = "chat_history"
PostgresChatMessageHistory.create_tables(sync_connection, table_name)

session_id = str(uuid.uuid4())

chat_history = PostgresChatMessageHistory(
    table_name,
    session_id,
    sync_connection=sync_connection
)

chat_history.add_messages([
    SystemMessage(content="Meow"),
    AIMessage(content="woof"),
    HumanMessage(content="bark"),
])

print(chat_history.messages)

PGVectorStore is the recommended class for vector storage (replacing the deprecated PGVector).

To use it:

1. Create a PGEngine from a connection string.
2. Initialize the vectorstore table using engine.init_vectorstore_table().
3. Create the store instance using PGVectorStore.create_sync() (or the asynchronous equivalent).

Note: All synchronous functions have corresponding asynchronous functions.

from langchain_core.documents import Document
from langchain_core.embeddings import DeterministicFakeEmbedding
from langchain_postgres import PGEngine, PGVectorStore

CONNECTION_STRING = "postgresql+psycopg://langchain:langchain@localhost:6024/langchain"
engine = PGEngine.from_connection_string(url=CONNECTION_STRING)

VECTOR_SIZE = 768
embedding = DeterministicFakeEmbedding(size=VECTOR_SIZE)

TABLE_NAME = "my_doc_collection"

engine.init_vectorstore_table(
    table_name=TABLE_NAME,
    vector_size=VECTOR_SIZE,
)

store = PGVectorStore.create_sync(
    engine=engine,
    table_name=TABLE_NAME,
    embedding_service=embedding,
)

docs = [
    Document(page_content="Apples and oranges"),
    Document(page_content="Cars and airplanes"),
    Document(page_content="Train")
]

store.add_documents(docs)

query = "I'd like a fruit."
results = store.similarity_search(query)
print(results)

Migrating from the legacy PGVector class to PGVectorStore moves your data from a two-table schema to a single-table-per-collection schema. This improves performance through better metadata handling (columns instead of JSON) and simplifies management.

Recommended Migration Workflow

1. Initialize the Engine: Create a PGEngine using your connection string.
2. Initialize Destination Table: Use engine.init_vectorstore_table to create the new table. You can customize metadata_columns and id_column for better filtering and non-UUID identifiers.
3. Create VectorStore Object: Instantiate PGVectorStore.create_sync (or the async equivalent) using the engine and your embedding service. If you customized columns in step 2, you must pass the same metadata_columns and id_column here.
4. Execute Migration: Use migrate_pgvector_collection to copy data. Setting delete_pg_collection=True will remove the data from the original PGVector tables after a successful copy.

from langchain_postgres import PGEngine, PGVectorStore
from langchain_postgres.utils.pgvector_migrator import migrate_pgvector_collection
from langchain_core.embeddings import FakeEmbeddings
from sqlalchemy import Column

# 1. Setup Engine
engine = PGEngine.from_connection_string(url="postgresql://user:pass@host/db")

# 2. Setup Destination Table
VECTOR_SIZE = 768
metadata_columns = [Column("col_0", "VARCHAR")]
engine.init_vectorstore_table(
    table_name="destination_table",
    vector_size=VECTOR_SIZE,
    metadata_columns=metadata_columns,
    id_column=Column("langchain_id", "VARCHAR"),
)

# 3. Create VectorStore
destination_vector_store = PGVectorStore.create_sync(
    engine,
    embedding_service=FakeEmbeddings(size=VECTOR_SIZE),
    table_name="destination_table",
    metadata_columns=["col_0"],
    id_column="langchain_id",
)

# 4. Migrate
migrate_pgvector_collection(
    engine,
    collection_name="my_collection",
    vector_store=destination_vector_store,
    delete_pg_collection=True,
)

A PGEngine is required to establish a connection to PostgreSQL. It manages a shared connection pool, which is a best practice for managing database connections and reducing latency.

PGVectorStore supports both asyncpg and psycopg3 drivers.

• To use the asyncpg driver, use a connection string starting with postgresql+asyncpg://.
• To use the psycopg3 driver, use a connection string starting with postgresql+psycopg://.

There are two ways to instantiate a PGEngine:

1. From a connection string: Use PGEngine.from_connection_string(url=...).
2. From an SQLAlchemy engine: Use PGEngine.from_engine(engine=...) with an AsyncEngine object.

from langchain_postgres import PGEngine
from sqlalchemy.ext.asyncio import create_async_engine

# Option 1: From connection string
CONNECTION_STRING = "postgresql+asyncpg://user:pass@host:port/db"
pg_engine = PGEngine.from_connection_string(url=CONNECTION_STRING)

# Option 2: From SQLAlchemy Async Engine
engine = create_async_engine(CONNECTION_STRING)
pg_engine = PGEngine.from_engine(engine=engine)

You can customize the schema of your vectorstore (e.g., renaming columns or using specific Postgres types for metadata) during table initialization and store creation.

Customizing the Table (ainit_vectorstore_table):

• content_column: Rename the text content column.
• embedding_column: Rename the vector column.
• metadata_columns: A list of Column objects to create dedicated relational columns for metadata (instead of just a JSON blob).
• id_column: A Column object to define a custom ID column (e.g., using INTEGER instead of UUID).

Customizing the Store (PGVectorStore.create):

• metadata_columns: Specify which relational columns should be treated as Document metadata.
• ignore_metadata_columns: Specify columns to ignore when mapping to Document metadata.
• distance_strategy: Change the similarity calculation method.
• index_query_options: Tune local index parameters for search.

from langchain_postgres import Column

# 1. Initialize table with a custom metadata column
await pg_engine.ainit_vectorstore_table(
    table_name="vectorstore_custom",
    vector_size=1024,
    metadata_columns=[Column("len", "INTEGER")],
)

# 2. Initialize store using those columns
custom_store = await PGVectorStore.create(
    engine=pg_engine,
    table_name="vectorstore_custom",
    embedding_service=embedding,
    metadata_columns=["len"],
)

When querying PGVector, you can pass a filter dictionary to restrict results based on the cmetadata JSONB field.

Supported Filter Syntax:

• Equality: {"field": "value"}
• Operators: Use a dictionary with a single key representing the operator.
  • $eq: Equality
  • $ne: Not equal
  • $gt: Greater than
  • $gte: Greater than or equal to
  • $lt: Less than
  • $lte: Less than or equal to
  • $between: {"field": {"$between": [low, high]}}
  • $in: {"field": {"$in": [val1, val2]}}
  • $nin: {"field": {"$nin": [val1, val2]}}
  • $like: Pattern matching
  • $ilike: Case-insensitive pattern matching
  • $exists: {"field": {"$exists": true/false}}
• Logical Operators:
  • $and: {"$and": [filter1, filter2, ...]}
  • $or: {"$or": [filter1, filter2, ...]}
  • $not: {"$not": filter} or {"$not": [filter1, filter2, ...]}

You can perform similarity searches using natural language queries or raw vectors. To ensure high performance, declare columns used for filtering in metadata_columns during initialization.

Supported Filter Operators:

• $eq, $ne: Equality, Inequality
• $lt, $lte, $gt, $gte: Comparison
• $in, $nin: Inclusion/Exclusion
• $between: Range
• $exists: Null check
• $like, $ilike: Text pattern matching (case-insensitive for $ilike)
• $and, $or: Logical operators

# Similarity search with a logical OR filter
await vectorstore.asimilarity_search(
    "birds", 
    filter={"$or": [{"topic": "animals"}, {"location": "market"}]}
)

# Similarity search with an IN filter
await vectorstore.asimilarity_search(
    "apple", 
    filter={"topic": {"$in": ["food", "animals"]}}
)

# Search by raw vector
query_vector = embedding.embed_query("fruit")
docs = await vectorstore.asimilarity_search_by_vector(query_vector, k=2)