LangChain integration

langchain-vectoramp provides a LangChain VectorStore, retriever, document loader, and an Intelligence (RAG) Runnable, all backed by VectorAmp datasets, VectorAmp-hosted embeddings, and SABLE.

VectorAmp embeds your text server-side, so you never wire up a LangChain Embeddings object: text add and search call VectorAmp's hosted model and SABLE index directly.

Install

pip install langchain-vectoramp

Set your API key once in the environment — every class reads it automatically:

export VECTORAMP_API_KEY="vsk_..."

Quickstart

from langchain_vectoramp import VectorAmpVectorStore

# API key comes from VECTORAMP_API_KEY; base_url defaults to production.
store = VectorAmpVectorStore(dataset_name="docs")   # or dataset_id="..."

store.add_texts(
    ["VectorAmp is built for billion-scale vector search."],
    metadatas=[{"source": "readme"}],
)

docs = store.similarity_search(
    "What is VectorAmp built for?",
    k=3,
    filter={"source": "readme"},
    rerank=True,                # expands to the VectorAmp-Rerank-v1 rerank object
)

add_texts embeds each text with the dataset's hosted model, copies the source text into metadata.text, generates ids when you omit them, and inserts the vectors in one call.

Embeddings are server-side

Every class accepts an optional LangChain Embeddings object (embedding=... on the vector store) for constructor compatibility only — it is intentionally ignored. Add and search always use VectorAmp's hosted model, so you never need to provide one.

Vector ids: strings or integers

ids accepts strings or integers. Integer ids are preserved and sent as JSON numbers, so the server stores exactly what you passed.

store.add_texts(["a", "b"], ids=[1, 2])      # -> [1, 2], stored as numbers
store.add_texts(["c"], ids=["doc-c"])        # -> ["doc-c"]

Retriever

as_retriever() is inherited from the LangChain VectorStore base class and works out of the box. VectorAmp-native search options pass through search_kwargs.

retriever = store.as_retriever(
    search_kwargs={"k": 5, "filter": {"tenant": "acme"}},
)
results = retriever.invoke("find relevant docs")

Search options

similarity_search and similarity_search_with_score (and their async a… variants) accept metadata filters and VectorAmp-native search kwargs:

docs = store.similarity_search(
    "renewal terms",
    k=5,
    filter={"source": "contracts"},   # LangChain convention (preferred)
    hybrid=True,                       # dense + sparse on hybrid datasets
    sparse_query="renewal",
    alpha=0.4,
    rerank=True,                       # model rerank (VectorAmp-Rerank-v1)
)

Filtering: pass filter={...} (the LangChain convention, preferred) or filters={...} (accepted for SDK parity) — use only one. Other passthrough kwargs: advanced_filters, include_metadata, include_documents, include_embeddings, embedding_provider, embedding_model, nprobe_override, rerank_depth_override.

rerank vs rerank_depth_override

These are two different knobs. rerank=True enables model reranking (VectorAmp-Rerank-v1) and reorders results by a cross-encoder. rerank_depth_override is a SABLE candidate-depth override for the vector index (how many candidates the index considers before scoring); it is not model rerank. Omit it to use the dataset default.

Intelligence (RAG answers)

VectorAmpVectorStore retrieves documents; VectorAmpIntelligence returns a generated answer with sources. The simplest usage needs only a question: it reads VECTORAMP_API_KEY, queries across all your datasets by default, and includes sources.

from langchain_vectoramp import VectorAmpIntelligence

intel = VectorAmpIntelligence()   # VECTORAMP_API_KEY, all datasets, sources on
print(intel.ask("What changed in the Q4 planning docs?"))

Multi-turn "just works": pass prior LangChain messages and they are forwarded as conversation history automatically. You decide how many previous turns to include simply by how many you pass.

from langchain_core.messages import HumanMessage, AIMessage

history = [
    HumanMessage("What is VectorAmp?"),
    AIMessage("A vector database platform."),
]
intel.ask("Does it support hybrid search?", history=history)

VectorAmpIntelligence is a Runnable, so it composes in LCEL chains and works with RunnableWithMessageHistory. Pass a list of messages and the last one is taken as the question; the rest become conversation history.

intel.invoke("Summarize the docs")
intel.invoke([HumanMessage("hi"), AIMessage("hello"), HumanMessage("and reranking?")])

# Need the sources/metadata too?
intel.ask_with_sources("What are the contract termination terms?")

Scope to one dataset or tune retrieval only when you need to:

intel = VectorAmpIntelligence(dataset_name="contracts", top_k=8)

Loader

VectorAmpLoader has two modes, selected by whether you pass query.

Retained original documents (no query): lists the dataset's retained source documents via GET /datasets/{id}/documents with cursor pagination, downloading originals where available.

from langchain_vectoramp import VectorAmpLoader

loader = VectorAmpLoader(
    dataset_id="dataset-id",
    metadata={"loaded_from": "vectoramp"},
)

documents = loader.load()
# or stream lazily:
for document in loader.lazy_load():
    print(document.metadata.get("id"), document.page_content[:200])

Search-backed (with query): returns the top k matching documents via semantic search.

loader = VectorAmpLoader(
    dataset_id="dataset-id",
    query="contract renewal terms",
    filter={"source": "contracts"},
    k=10,
)
documents = loader.load()

Notes:

• Documents without retained originals are skipped rather than exposing storage internals.
• Downloads follow redirect behavior from the API.
• Public document metadata is stored on Document.metadata; storage buckets, object keys, and payload refs are never exposed.

Method reference

VectorAmpVectorStore

Constructor (keyword-only):

VectorAmpVectorStore(
    *,
    api_key: str | None = None,        # default: env VECTORAMP_API_KEY
    base_url: str = "https://api.vectoramp.com",
    dataset_id: str | None = None,     # provide exactly one of dataset_id / dataset_name
    dataset_name: str | None = None,   # resolved to an id via GET /datasets
    client: VectorAmpHTTPClient | None = None,
    embedding: Embeddings | None = None,   # accepted but IGNORED (server-side embedding)
    timeout: float = 30.0,
)

Method	Required	Optional (defaults)	Returns
add_texts(texts, metadatas=None, *, ids=None, **kwargs)	texts	metadatas, ids (auto UUIDs; str or int)	list[str | int] ids
aadd_texts(texts, metadatas=None, *, ids=None, **kwargs)	texts	metadatas, ids	list[str | int]
similarity_search(query, k=4, **kwargs)	query	k (4), filter/filters, hybrid, sparse_query, alpha, rerank, advanced_filters, include_metadata, include_documents, include_embeddings, embedding_provider, embedding_model, nprobe_override, rerank_depth_override	list[Document]
similarity_search_with_score(query, k=4, **kwargs)	query	same as above	list[tuple[Document, float]]
asimilarity_search(query, k=4, **kwargs)	query	same as above	list[Document]
asimilarity_search_with_score(query, k=4, **kwargs)	query	same as above	list[tuple[Document, float]]
as_retriever(**kwargs) (inherited)	—	search_kwargs (e.g. {"k": 5, "filter": {...}}), search_type	VectorStoreRetriever
add_documents(documents, **kwargs) (inherited)	documents	ids	list[str]
from_texts(texts, embedding=None, metadatas=None, *, ids=None, **kwargs) (classmethod)	texts, plus dataset_id or dataset_name in kwargs	embedding (ignored), metadatas, ids	VectorAmpVectorStore
from_documents(documents, embedding=None, **kwargs) (classmethod)	documents, plus dataset_id or dataset_name	embedding (ignored)	VectorAmpVectorStore
afrom_texts_instance(texts, metadatas=None, *, ids=None)	texts	metadatas, ids	list[str | int]
close() / aclose()	—	—	None

Properties: dataset_id (resolved id), dataset_name (original name, if given), embeddings (returns the unused embedding arg).

VectorAmpIntelligence

Constructor (keyword-only):

VectorAmpIntelligence(
    *,
    dataset_id: str | None = None,     # default: query across all datasets ("all")
    dataset_name: str | None = None,
    api_key: str | None = None,        # default: env VECTORAMP_API_KEY
    base_url: str = "https://api.vectoramp.com",
    client: VectorAmpHTTPClient | None = None,
    top_k: int | None = None,          # default: server default (5)
    include_sources: bool = True,
)

Method	Required	Optional (defaults)	Returns
ask(query, *, history=None, **overrides)	query	history, and overrides dataset_id, include_sources, top_k	str answer
aask(query, *, history=None, **overrides)	query	same as ask	str
ask_with_sources(query, *, history=None, **overrides)	query	same as ask	dict (answer, sources, metadata)
invoke(input, config=None, **kwargs)	input	—	str
ainvoke(input, config=None, **kwargs)	input	—	str

invoke/ainvoke accept a plain string, a dict ({query|question|input, history|chat_history|messages}), or a list of LangChain messages (the last is the question, the rest become conversation history). Message types map to API roles: human → user, ai → assistant, system → system, tool → tool.

VectorAmpLoader

Constructor (keyword-only):

VectorAmpLoader(
    *,
    dataset_id: str,                   # required
    api_key: str | None = None,        # default: env VECTORAMP_API_KEY
    base_url: str = "https://api.vectoramp.com",
    client: VectorAmpHTTPClient | None = None,
    query: str | None = None,          # omit -> list documents; provide -> semantic search
    filter: Mapping | None = None,
    k: int = 10,
    metadata: Mapping | None = None,   # merged into every loaded Document's metadata
    timeout: float = 30.0,
    **search_kwargs,                   # forwarded to search when query is set
)

Method	Returns	Behavior
load()	list[Document]	Materializes all documents
lazy_load()	Iterator[Document]	Streams documents (paginates the documents API, or runs one search when query is set)
close()	None	Closes owned HTTP resources

Retry ingestion jobs

The package's underlying HTTP client exposes job-management helpers. Retry an eligible failed or cancelled ingestion job as a fresh full rerun:

store._client.retry_job("job-uuid")          # sync
await store._client.aretry_job("job-uuid")    # async

Authenticated sources & cleanup

The same underlying client exposes ingestion source and connection management, mirroring the Python SDK. For example, ingest an authenticated Google Drive folder into the store's dataset — either headless with a service account, or via a reusable connection granted once in the browser:

# Option 1 — service account (headless, no browser).
src = store._client.create_source(
    source_type="gdrive",
    name="Team Drive",
    config={
        "folder_ids": ["drive-folder-id"],
        "service_account_json": {"type": "service_account", "client_email": "ingest@proj.iam.gserviceaccount.com", "private_key": "…"},
    },
)
store._client.start_job(source_id=src["id"], dataset_id=store.dataset_id)

# Option 2 — connection (interactive once, reused server-side).
conn = store._client.create_connection(provider="google", source_type="gdrive")
print("Open in a browser:", conn["authorization_url"])
# poll store._client.get_connection(conn["id"]) until status == "connected", then:
store._client.create_source(
    source_type="gdrive",
    name="Team Drive",
    config={"folder_ids": ["drive-folder-id"], "connection_id": conn["id"]},
)

Tidy up unused sources the same way:

store._client.delete_source("source-uuid", force=True)   # 409 unless force (force disables its schedules)
store._client.cleanup_unused_sources()                   # remove every unused source

See the Sources and Connections API references for the full surface.

Development

pip install -e '.[dev]'
pytest
ruff check .
mypy

Tests use mocked clients/transports only; no live VectorAmp API calls are made.

License

Apache License 2.0. See LICENSE and NOTICE.