AI Engineer

5 minute read

Building Production-Ready Vector Search Applications with TimescaleVector and LangChain: A PostgreSQL-Based Approach

Vector search has become an essential component of modern AI applications, enabling semantic search capabilities and powering retrieval-augmented generation (RAG) systems. While many vector database options exist, PostgreSQL-based solutions offer a compelling advantage for teams that already have PostgreSQL infrastructure in their stack. In this article, we’ll explore how to implement efficient vector search using TimescaleVector with LangChain, a powerful combination that leverages the reliability of PostgreSQL while providing specialized vector search capabilities.

TimescaleVector is a PostgreSQL extension that adds vector similarity search capabilities to your existing PostgreSQL database. This approach offers several advantages:

  1. Unified infrastructure: Keep your vector data alongside your relational data
  2. Familiar tooling: Use standard PostgreSQL management tools
  3. SQL integration: Combine vector search with complex SQL queries
  4. Enterprise-grade reliability: Benefit from PostgreSQL’s robustness

For teams already using PostgreSQL, TimescaleVector provides a seamless path to implementing vector search without adding a separate database to your stack.

Getting Started with TimescaleVector in LangChain

LangChain provides a convenient TimescaleVector class that simplifies working with TimescaleVector. Let’s start by looking at the basic setup.

Installation

First, install the required packages:

pip install langchain timescale_vector psycopg2-binary

Basic Configuration

To use TimescaleVector with LangChain, you’ll need to provide a connection string and an embedding model:

from langchain_community.vectorstores import TimescaleVector
from langchain_openai import OpenAIEmbeddings

# Initialize embedding model
embeddings = OpenAIEmbeddings()

# Connect to TimescaleVector
service_url = "postgres://username:password@hostname:port/database"
vector_store = TimescaleVector(
    service_url=service_url,
    embedding=embeddings,
    collection_name="my_document_collection"
)

You can also use environment variables for the connection string:

import os
os.environ["TIMESCALE_SERVICE_URL"] = "postgres://username:password@hostname:port/database"

# Now you can initialize without explicitly providing the service_url
vector_store = TimescaleVector(
    embedding=embeddings,
    collection_name="my_document_collection"
)

Adding Documents to TimescaleVector

Once you’ve configured your vector store, you can add documents to it. LangChain provides several methods for this purpose.

Adding Documents from Texts

The simplest approach is to add documents from raw text:

texts = [
    "TimescaleVector provides efficient vector search in PostgreSQL",
    "LangChain simplifies building AI applications with LLMs",
    "Vector search enables semantic similarity queries",
    "PostgreSQL is a powerful open-source relational database"
]

# Optional metadata for each document
metadatas = [
    {"source": "docs", "category": "database"},
    {"source": "blog", "category": "ai"},
    {"source": "tutorial", "category": "search"},
    {"source": "docs", "category": "database"}
]

# Add texts to the vector store
ids = vector_store.add_texts(texts=texts, metadatas=metadatas)

Creating from Existing Documents

If you already have LangChain Document objects, you can create a vector store directly:

from langchain_core.documents import Document

documents = [
    Document(page_content="Document 1 content", metadata={"source": "file1.txt"}),
    Document(page_content="Document 2 content", metadata={"source": "file2.txt"})
]

# Create vector store from documents
vector_store = TimescaleVector.from_documents(
    documents=documents,
    embedding=embeddings,
    collection_name="my_documents"
)

Working with Pre-computed Embeddings

For performance optimization, you might want to pre-compute embeddings:

# Pre-compute embeddings
text_embeddings = [
    ("Document 1 content", embeddings.embed_query("Document 1 content")),
    ("Document 2 content", embeddings.embed_query("Document 2 content"))
]

# Create vector store from pre-computed embeddings
vector_store = TimescaleVector.from_embeddings(
    text_embeddings=text_embeddings,
    embedding=embeddings,
    collection_name="precomputed_embeddings"
)

TimescaleVector provides several search methods to retrieve relevant documents.

The most common search operation is a basic similarity search:

query = "How does vector search work in PostgreSQL?"
docs = vector_store.similarity_search(query, k=3)

for doc in docs:
    print(doc.page_content)
    print(doc.metadata)
    print("---")

Search with Metadata Filtering

You can filter search results based on metadata:

# Search only in documents with category "database"
filter_dict = {"category": "database"}
docs = vector_store.similarity_search(
    query="PostgreSQL features",
    k=3,
    filter=filter_dict
)

Search with Scores

To get similarity scores along with the documents:

results = vector_store.similarity_search_with_score(
    query="vector similarity",
    k=3
)

for doc, score in results:
    print(f"Content: {doc.page_content}")
    print(f"Score: {score}")
    print("---")

For more diverse search results, you can use MMR search:

docs = vector_store.max_marginal_relevance_search(
    query="database technologies",
    k=3,
    fetch_k=10,
    lambda_mult=0.5  # Controls diversity (0 = max diversity, 1 = min diversity)
)

Advanced Features

TimescaleVector in LangChain provides several advanced features for production applications.

Indexing

For better performance, you should create an index:

# Create a default index
vector_store.create_index()

# Or specify the index type
from langchain_community.vectorstores.timescalevector import IndexType
vector_store.create_index(index_type=IndexType.HNSW)

Time-Based Partitioning

TimescaleVector supports time-based partitioning, which is useful for managing large datasets:

from datetime import timedelta

vector_store = TimescaleVector(
    service_url=service_url,
    embedding=embeddings,
    collection_name="time_partitioned_docs",
    time_partition_interval=timedelta(days=30)  # Create new partitions every 30 days
)

Document Management

You can also manage documents in your vector store:

# Delete documents by ID
vector_store.delete(ids=["doc_id_1", "doc_id_2"])

# Delete documents by metadata
vector_store.delete_by_metadata(filter={"category": "outdated"})

Building a Retriever

TimescaleVector integrates seamlessly with LangChain’s retriever interface, making it easy to use in RAG applications:

retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

# Use in a RAG chain
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA

llm = ChatOpenAI(model_name="gpt-4")
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever
)

result = qa_chain.invoke("What are the advantages of using TimescaleVector?")
print(result["result"])

Asynchronous Operations

TimescaleVector supports asynchronous operations, which can be crucial for high-throughput applications:

import asyncio

async def async_search_demo():
    query = "vector search in PostgreSQL"
    docs = await vector_store.asimilarity_search(query, k=3)
    return docs

# Run the async function
docs = asyncio.run(async_search_demo())

Production Considerations

When deploying a TimescaleVector-based solution to production, consider these best practices:

  1. Connection pooling: Use connection pooling to manage database connections efficiently
  2. Index optimization: Choose the right index type for your use case
  3. Monitoring: Set up monitoring for query performance and database health
  4. Backup strategy: Implement regular backups of your vector data
  5. Scaling: Consider read replicas for scaling read operations
# Example of using a custom logger for monitoring
import logging

logger = logging.getLogger("timescale_vector")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
logger.addHandler(handler)

vector_store = TimescaleVector(
    service_url=service_url,
    embedding=embeddings,
    collection_name="production_vectors",
    logger=logger
)

Conclusion

TimescaleVector with LangChain provides a powerful, PostgreSQL-based approach to implementing vector search in your applications. By leveraging the familiarity and reliability of PostgreSQL while adding specialized vector search capabilities, this combination offers an excellent solution for teams that want to add semantic search to their existing PostgreSQL infrastructure.

This approach is particularly valuable for organizations that:

  • Already use PostgreSQL in their stack
  • Need to combine vector search with complex SQL queries
  • Want to minimize the number of database systems they maintain
  • Require enterprise-grade reliability for their vector search solution

By following the patterns outlined in this article, you can build production-ready vector search applications that seamlessly integrate with your existing data infrastructure.

This post was originally written in my native language and then translated using an LLM. I apologize if there are any grammatical inconsistencies.

Categories:

Updated:

You may also enjoy