LangChain

Applies to VAST Cluster 5.5. Examples use the VAST DataBase Python SDK 2.1.0.

langchain-vastdb is a LangChain integration for VAST DataBase. Its VastDBVectorStore class is a LangChain vector store backed by a VAST DataBase table: you add documents, search them, and hand the store to a RAG chain as a retriever, with the usual LangChain calls. Documents are written to the table with the Python SDK, and searches, lookups and deletes run as SQL in the VAST Query Engine, through ADBC.

The package is open source (Apache-2.0), on PyPI and GitHub. It's in alpha, so its API can change between releases: pin the version you test with. This page uses langchain-vastdb 0.0.6.

Run the code blocks in order in one Python session. Below each block is the output you should see.

Before you start

You need what any Vector Search client needs: a database, an access key pair with s3:Tabular* and s3:QueryEngine*, and the Query Engine endpoint. Setup has the list to send to your platform team.

Install the integration, the SDK and the ADBC driver, and an embedding library. This page uses FastEmbed, which runs a small embedding model on your CPU:

pip install langchain-vastdb==0.0.6 vastdb==2.1.0 adbc-driver-vastdb adbc-driver-manager fastembed

Set QE_ENDPOINT, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and VASTDB_BUCKET, as in setup. The store and the SDK read the keys from the two AWS_ variables.

Choose an embedding model

The store takes any LangChain Embeddings object, such as OpenAIEmbeddings or HuggingFaceEmbeddings. The example wraps FastEmbed's BAAI/bge-small-en-v1.5 model, which makes 384-dimensional vectors and downloads on first use:

from fastembed import TextEmbedding
from langchain_core.embeddings import Embeddings

class FastEmbedEmbeddings(Embeddings):
    def __init__(self, model_name="BAAI/bge-small-en-v1.5"):
        self.model = TextEmbedding(model_name)

    def embed_documents(self, texts):
        return [v.tolist() for v in self.model.embed(texts)]

    def embed_query(self, text):
        return self.embed_documents([text])[0]

embeddings = FastEmbedEmbeddings()
DIM = len(embeddings.embed_query("hello"))
print(DIM)

The table's vector column must have the same dimension as the model's vectors.

Create the table

The store doesn't create a vector index. Create the table yourself with the Python SDK and a VectorIndexSpec, then point the store at it. The store expects four columns, with these default names: id and text (strings), embedding (a fixed-size list of float32) and metadata (a string, which holds each document's metadata as JSON). Add a column for each metadata field you want to filter on, here source:

import os
import adbc_driver_vastdb
import pyarrow as pa
import vastdb
from vastdb import VectorIndexSpec

BUCKET = os.environ["VASTDB_BUCKET"]
SCHEMA = "lc_docs"

session = vastdb.connect(endpoint=os.environ["QE_ENDPOINT"])
vector = pa.list_(pa.field("item", pa.float32(), nullable=False), DIM)
columns = pa.schema([
    ("id", pa.string()),
    ("text", pa.string()),
    ("embedding", vector),
    ("metadata", pa.string()),
    ("source", pa.string()),
])

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).create_schema(SCHEMA, fail_if_exists=False)
    table = schema.create_table(
        "kb",
        columns,
        vector_index=VectorIndexSpec(column="embedding", distance_metric="cosine"),
    )
    print(table.vector_index.distance_metric, table.vector_index.sql_distance_function)

cosine array_cosine_distance

The store reads the metric from the index, so you don't pass one. build_table() and build_with_table(), the store's own helpers, create a table without an index; see tables without an index.

Open the store

A store that should filter on source declares it as a typed metadata column, in a subclass. Give it the session, the table's location, and the ADBC driver and Query Engine endpoint:

from langchain_vastdb import TypedColumn, VastDBVectorStore

class KnowledgeBase(VastDBVectorStore):
    _typed_metadata_columns = {"source": TypedColumn()}

store = KnowledgeBase(
    embedding=embeddings,
    session=session,
    bucket=BUCKET,
    schema=SCHEMA,
    table_name="kb",
    adbc_driver_path=adbc_driver_vastdb.get_driver_path(),
    adbc_endpoint=os.environ["QE_ENDPOINT"],
)

Without adbc_driver_path and adbc_endpoint, get_by_ids() still works, through the SDK, but searches fail; see errors.

Add documents

add_texts() embeds the texts with your model and inserts them, with their metadata. Pass your own IDs to keep them stable, or leave them out and the store generates one per document:

texts = [
    "Paris is the capital of France.",
    "Berlin is the capital of Germany.",
    "The Eiffel Tower is in Paris.",
    "Bananas are rich in potassium.",
    "VAST DataBase stores tables and vectors together.",
]
metadatas = [
    {"source": "geo", "lang": "en"},
    {"source": "geo", "lang": "en"},
    {"source": "travel", "lang": "en"},
    {"source": "food", "lang": "en"},
    {"source": "docs", "lang": "en"},
]
ids = store.add_texts(texts, metadatas=metadatas, ids=[f"doc-{i}" for i in range(1, 6)])
print(ids)

['doc-1', 'doc-2', 'doc-3', 'doc-4', 'doc-5']

source goes into its own column. Metadata fields without a column of their own, here lang, go into the metadata column as JSON.

Search

similarity_search() embeds the question and returns the nearest documents, as LangChain Document objects:

for doc in store.similarity_search("What is the capital of France?", k=2):
    print(doc.id, doc.page_content, doc.metadata)

doc-1 Paris is the capital of France. {'lang': 'en', 'source': 'geo'}
doc-3 The Eiffel Tower is in Paris. {'lang': 'en', 'source': 'travel'}

similarity_search_with_score() also returns each document's distance. Lower is closer: for a cosine index, the score is the cosine distance.

for doc, score in store.similarity_search_with_score("capital city", k=3):
    print(f"{score:.3f}", doc.page_content)

0.308 Berlin is the capital of Germany.
0.331 Paris is the capital of France.
0.418 The Eiffel Tower is in Paris.

To search with a vector you already have, use similarity_search_by_vector().

Use the store as a retriever

as_retriever() turns the store into a LangChain retriever, the piece a RAG chain calls for context:

retriever = store.as_retriever(search_kwargs={"k": 2})
for doc in retriever.invoke("Where is the Eiffel Tower?"):
    print(doc.page_content)

The Eiffel Tower is in Paris.
Paris is the capital of France.

In a chain, the retriever goes where the context comes from. This fragment needs an LLM of your own:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

prompt = ChatPromptTemplate.from_template("Answer from this context:\n{context}\n\nQuestion: {question}")
chain = (
    {"context": retriever | (lambda docs: "\n".join(d.page_content for d in docs)),
     "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)
print(chain.invoke("Where is the Eiffel Tower?"))

Filter on metadata

Pass an ibis expression as predicate. The store turns it into a SQL WHERE clause, so it can only name columns: here source, not lang, which is inside the JSON. Write column names as _["source"]:

from ibis import _

for doc in store.similarity_search("Paris", k=3, predicate=_["source"] == "geo"):
    print(doc.page_content)
print()
for doc in store.similarity_search("Paris", k=3, predicate=_["source"].isin(["travel", "food"])):
    print(doc.page_content)

Paris is the capital of France.
Berlin is the capital of Germany.

The Eiffel Tower is in Paris.
Bananas are rich in potassium.

You can combine conditions with & (and) and | (or), and compare with ==, !=, <, <=, > and >=. The other forms fail:

for predicate in (_.source == "geo", _["lang"] == "en"):
    try:
        store.similarity_search("Paris", k=3, predicate=predicate)
    except Exception as e:
        print(type(e).__name__ + ":", str(e).splitlines()[0][:100])

TypeError: Unsupported ibis resolver node type 'Attr': _.source
InternalError: INTERNAL: AdbcStatementExecute failed to start query

_.source isn't supported by the store's filter translation; use _["source"]. A field that has no column of its own, such as lang, makes the Query Engine reject the query. The older filter= keyword argument was removed from the package and raises a TypeError that says so.

Look up, replace and delete documents

print([d.page_content for d in store.get_by_ids(["doc-1", "doc-4"])])

store.add_texts(["Paris is the capital and largest city of France."],
                metadatas=[{"source": "geo", "lang": "en"}], ids=["doc-1"])
print([d.page_content for d in store.get_by_ids(["doc-1"])])

store.delete(ids=["doc-4"])
print([d.id for d in store.similarity_search("fruit", k=5)])

['Paris is the capital of France.', 'Bananas are rich in potassium.']
['Paris is the capital and largest city of France.']
['doc-5', 'doc-2', 'doc-1', 'doc-3']

Adding a document with an ID that's already in the table replaces it: when you pass IDs, the store first deletes the rows with those IDs, in the same transaction as the insert. get_by_ids() skips IDs it doesn't find.

Count documents

count() without arguments reads the table's statistics, which the cluster updates in the background, so it can lag recent writes. Pass a predicate for an exact count:

print(store.count())
print(store.count(_["source"] == "geo"))

0
2

How it runs on VAST

Call	Runs as
`add_texts()`	An insert through the Python SDK, in a transaction. If you pass IDs, a Query Engine `DELETE` of those IDs runs first, joined to the same transaction.
`similarity_search()` and the other searches	One Query Engine query through ADBC: `SELECT` the columns and the distance, `WHERE` your predicate, `ORDER BY` the distance, `LIMIT` k
`get_by_ids()`	A Query Engine `SELECT … WHERE "id" = …`
`delete()`	A Query Engine `DELETE … WHERE "id" = …`

A search on the table above sends this SQL, with the query vector in place of ARRAY[...]:

SELECT "id", "text", "source", "metadata", array_cosine_distance("embedding"::FLOAT[384], ARRAY[...]::FLOAT[384]) AS "_vastdb_distance" FROM "<bucket>/lc_docs"."kb" WHERE "source" = 'geo' ORDER BY "_vastdb_distance" LIMIT 3

The distance function is the one the index's metric maps to: array_cosine_distance for cosine, array_distance for l2sq, and array_inner_product for ip. With an ip index, the store negates the inner product so that lower scores are closer, as for the other metrics: an ip score is negative, and the best match has the most negative score.

Searches on a cosine or l2sq index have the shape the vector index answers, with or without a predicate: on a table big enough for the index to be built, EXPLAIN of the store's query shows a VAST_VEC_INDEX_SCAN step. Searches on an ip index don't use the index; they compare the query vector with every row, so prefer cosine or l2sq for large tables. On a new or small table, searches scan every row until the index is built, as for any client; see query design.

The store keeps one ADBC connection per thread and reconnects after an error. It loads the table's metadata, including the index definition, on first use; if you change the table from elsewhere, call store.invalidate_table_cache().

Tables without an index

build_table() and build_with_table() create the schema and a table with the store's columns, but no vector index. On such a table:

You must pass distance_metric= ("l2sq", "cosine" or "ip") when you open the store. Without it, the first call fails with ValueError: VastDB cluster returned no vector index metadata for table … Pass distance_metric= to the constructor.
distance_metric="cosine" doesn't work: the store calls a cosine_distance function, which the VAST Query Engine doesn't have, so every search fails. Use a table with a cosine index, or "l2sq". For unit-length vectors, which many embedding models produce, l2sq ranks results in the same order as cosine.
Every search compares the query vector with every row.

Errors

from langchain_core.embeddings import DeterministicFakeEmbedding

no_adbc = KnowledgeBase(embedding=embeddings, session=session, bucket=BUCKET, schema=SCHEMA, table_name="kb")
wrong_dim = KnowledgeBase(embedding=DeterministicFakeEmbedding(size=8), session=session, bucket=BUCKET,
                          schema=SCHEMA, table_name="kb", adbc_driver_path=adbc_driver_vastdb.get_driver_path(),
                          adbc_endpoint=os.environ["QE_ENDPOINT"])
for label, call in (("no ADBC", lambda: no_adbc.similarity_search("Paris", k=1)),
                    ("wrong dimension", lambda: wrong_dim.add_texts(["x"]))):
    try:
        call()
    except Exception as e:
        detail = f"{e.code}: {e.message}" if hasattr(e, "code") else str(e)[:120]
        print(label, "->", type(e).__name__ + ":", detail)

no ADBC -> RuntimeError: ADBC is not configured. Vector search requires a Query Engine (ADBC) connection. Provide adbc_driver_path, adbc_endpoint
wrong dimension -> BadRequest: TabularMismatchColumnType: Mismatched column type between file to import and existing table.

No ADBC: searches need the Query Engine. The error also mentions VASTDB_ALLOW_FALLBACK=1, which makes the store read the whole table into your process and search it there; the package says that's for development and testing only.
Wrong dimension: the model's vectors don't match the table's vector column. The SDK rejects the insert with TabularMismatchColumnType, and nothing is written.

Clean up

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).schema(SCHEMA)
    schema.table("kb").drop()
    schema.drop()
print("Dropped", SCHEMA)

Dropped lc_docs