langchain_couchbase package

Submodules

langchain_couchbase.cache module

LangChain Couchbase Caches

Functions “_hash”, “_loads_generations” and “_dumps_generations” are duplicated in this utility from modules:

• “libs/community/langchain_community/cache.py”

class langchain_couchbase.cache.CouchbaseCache(cluster: Cluster, bucket_name: str, scope_name: str, collection_name: str, ttl: timedelta | None = None, **kwargs: Dict[str, Any])

Bases: BaseCache

Couchbase LLM Cache LLM Cache that uses Couchbase as the backend

LLM = 'llm'

PROMPT = 'prompt'

RETURN_VAL = 'return_val'

clear(**kwargs: Any) → None

Clear the cache. This will delete all documents in the collection. This requires an index on the collection.

lookup(prompt: str, llm_string: str) → Sequence[Generation] | None

Look up from cache based on prompt and llm_string.

update(prompt: str, llm_string: str, return_val: Sequence[Generation]) → None

Update cache based on prompt and llm_string.

class langchain_couchbase.cache.CouchbaseSemanticCache(cluster: Cluster, embedding: Embeddings, bucket_name: str, scope_name: str, collection_name: str, index_name: str, score_threshold: float | None = None, ttl: timedelta | None = None)

Bases: BaseCache, CouchbaseVectorStore

Couchbase Semantic Cache Cache backed by a Couchbase Server with Vector Store support

LLM = 'llm_string'

RETURN_VAL = 'return_val'

clear(**kwargs: Any) → None

Clear the cache. This will delete all documents in the collection. This requires an index on the collection.

lookup(prompt: str, llm_string: str) → Sequence[Generation] | None

Look up from cache based on the semantic similarity of the prompt

update(prompt: str, llm_string: str, return_val: Sequence[Generation]) → None

Update cache based on the prompt and llm_string

langchain_couchbase.chat_message_histories module

Couchbase Chat Message History

class langchain_couchbase.chat_message_histories.CouchbaseChatMessageHistory(*, cluster: Cluster, bucket_name: str, scope_name: str, collection_name: str, session_id: str, session_id_key: str = 'session_id', message_key: str = 'message', create_index: bool = True, ttl: timedelta | None = None)

Bases: BaseChatMessageHistory

Couchbase Chat Message History Chat message history that uses Couchbase as the storage

add_message(message: BaseMessage) → None

Add a message to the cache

add_messages(messages: Sequence[BaseMessage]) → None

Add messages to the cache in a batched manner

clear() → None

Clear the cache

property messages: List[BaseMessage]

Get all messages in the cache associated with the session_id

langchain_couchbase.vectorstores module

Couchbase vector stores.

class langchain_couchbase.vectorstores.CouchbaseVectorStore(cluster: Cluster, bucket_name: str, scope_name: str, collection_name: str, embedding: Embeddings, index_name: str, *, text_key: str | None = 'text', embedding_key: str | None = 'embedding', scoped_index: bool = True)

Bases: VectorStore

__ModuleName__ vector store integration.

Setup:

Install langchain-couchbase and head over to the Couchbase [website](https://cloud.couchbase.com) and create a new connection, with a bucket, collection, and search index.

pip install -U langchain-couchbase

import getpass

COUCHBASE_CONNECTION_STRING = getpass.getpass("Enter the connection string for the Couchbase cluster: ")
DB_USERNAME = getpass.getpass("Enter the username for the Couchbase cluster: ")
DB_PASSWORD = getpass.getpass("Enter the password for the Couchbase cluster: ")

Key init args — indexing params:

embedding: Embeddings

Embedding function to use.

Key init args — client params:

cluster: Cluster

Couchbase cluster object with active connection.

bucket_name: str

Name of the bucket to store documents in.

scope_name: str

Name of the scope in the bucket to store documents in.

collection_name: str

Name of the collection in the scope to store documents in.

index_name: str

Name of the Search index to use.

Instantiate:

from datetime import timedelta
from langchain_openai import OpenAIEmbeddings
from couchbase.auth import PasswordAuthenticator
from couchbase.cluster import Cluster
from couchbase.options import ClusterOptions

auth = PasswordAuthenticator(DB_USERNAME, DB_PASSWORD)
options = ClusterOptions(auth)
cluster = Cluster(COUCHBASE_CONNECTION_STRING, options)

# Wait until the cluster is ready for use.
cluster.wait_until_ready(timedelta(seconds=5))

BUCKET_NAME = "langchain_bucket"
SCOPE_NAME = "_default"
COLLECTION_NAME = "default"
SEARCH_INDEX_NAME = "langchain-test-index"

vector_store = CouchbaseVectorStore(
    cluster=cluster,
    bucket_name=BUCKET_NAME,
    scope_name=SCOPE_NAME,
    collection_name=COLLECTION_NAME,
    embedding=embeddings,
    index_name=SEARCH_INDEX_NAME,
)

Add Documents:

from langchain_core.documents import Document

document_1 = Document(page_content="foo", metadata={"baz": "bar"})
document_2 = Document(page_content="thud", metadata={"bar": "baz"})
document_3 = Document(page_content="i will be deleted :(")

documents = [document_1, document_2, document_3]
ids = ["1", "2", "3"]
vector_store.add_documents(documents=documents, ids=ids)

Delete Documents:

vector_store.delete(ids=["3"])

# TODO: Fill out with example output. Search:

results = vector_store.similarity_search(query="thud",k=1)
for doc in results:
    print(f"* {doc.page_content} [{doc.metadata}]")

# TODO: Example output

# TODO: Fill out with relevant variables and example output. Search with filter:

# TODO: Update filter to correct format
results = vector_store.similarity_search(query="thud",k=1,filter={"bar": "baz"})
for doc in results:
    print(f"* {doc.page_content} [{doc.metadata}]")

# TODO: Example output

# TODO: Fill out with example output. Search with score:

results = vector_store.similarity_search_with_score(query="qux",k=1)
for doc, score in results:
    print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")

# TODO: Example output

# TODO: Fill out with example output. Async:

# add documents
# await vector_store.aadd_documents(documents=documents, ids=ids)

# delete documents
# await vector_store.adelete(ids=["3"])

# search
# results = vector_store.asimilarity_search(query="thud",k=1)

# search with score
results = await vector_store.asimilarity_search_with_score(query="qux",k=1)
for doc,score in results:
    print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")

# TODO: Example output

# TODO: Fill out with example output. Use as Retriever:

retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 1, "fetch_k": 2, "lambda_mult": 0.5},
)
retriever.invoke("thud")

# TODO: Example output

DEFAULT_BATCH_SIZE = 100

add_texts(texts: Iterable[str], metadatas: List[dict] | None = None, ids: List[str] | None = None, batch_size: int | None = None, **kwargs: Any) → List[str]

Run texts through the embeddings and persist in vectorstore.

If the document IDs are passed, the existing documents (if any) will be overwritten with the new ones.

Args:

texts (Iterable[str]): Iterable of strings to add to the vectorstore. metadatas (Optional[List[Dict]]): Optional list of metadatas associated

with the texts.

ids (Optional[List[str]]): Optional list of ids associated with the texts.

IDs have to be unique strings across the collection. If it is not specified uuids are generated and used as ids.

batch_size (Optional[int]): Optional batch size for bulk insertions.

Default is 100.

Returns:

List[str]:List of ids from adding the texts into the vectorstore.

delete(ids: List[str] | None = None, **kwargs: Any) → bool | None

Delete documents from the vector store by ids.

Args:

ids (List[str]): List of IDs of the documents to delete. batch_size (Optional[int]): Optional batch size for bulk deletions.

Returns:

bool: True if all the documents were deleted successfully, False otherwise.

property embeddings: Embeddings

Return the query embedding object.

classmethod from_texts(texts: List[str], embedding: Embeddings, metadatas: List[dict] | None = None, **kwargs: Any) → CouchbaseVectorStore

Construct a Couchbase vector store from a list of texts.

Example:

from langchain_couchbase import CouchbaseVectorStore from langchain_openai import OpenAIEmbeddings

from couchbase.cluster import Cluster from couchbase.auth import PasswordAuthenticator from couchbase.options import ClusterOptions from datetime import timedelta

auth = PasswordAuthenticator(username, password) options = ClusterOptions(auth) connect_string = “couchbases://localhost” cluster = Cluster(connect_string, options)

# Wait until the cluster is ready for use. cluster.wait_until_ready(timedelta(seconds=5))

embeddings = OpenAIEmbeddings()

texts = [“hello”, “world”]

vectorstore = CouchbaseVectorStore.from_texts(

texts, embedding=embeddings, cluster=cluster, bucket_name=””, scope_name=””, collection_name=””, index_name=”vector-index”,

)

Args:

texts (List[str]): list of texts to add to the vector store. embedding (Embeddings): embedding function to use. metadatas (optional[List[Dict]): list of metadatas to add to documents. **kwargs: Keyword arguments used to initialize the vector store with and/or

passed to add_texts method. Check the constructor and/or add_texts for the list of accepted arguments.

Returns:

A Couchbase vector store.

similarity_search(query: str, k: int = 4, search_options: Dict[str, Any] | None = {}, **kwargs: Any) → List[Document]

Return documents most similar to embedding vector with their scores.

Args:

query (str): Query to look up for similar documents k (int): Number of Documents to return.

Defaults to 4.

search_options (Optional[Dict[str, Any]]): Optional search options that are

passed to Couchbase search. Defaults to empty dictionary

fields (Optional[List[str]]): Optional list of fields to include in the

metadata of results. Note that these need to be stored in the index. If nothing is specified, defaults to all the fields stored in the index.

Returns:

List of Documents most similar to the query.

similarity_search_by_vector(embedding: List[float], k: int = 4, search_options: Dict[str, Any] | None = {}, **kwargs: Any) → List[Document]

Return documents that are most similar to the vector embedding.

Args:

embedding (List[float]): Embedding to look up documents similar to. k (int): Number of Documents to return.

Defaults to 4.

search_options (Optional[Dict[str, Any]]): Optional search options that are

passed to Couchbase search. Defaults to empty dictionary.

fields (Optional[List[str]]): Optional list of fields to include in the

metadata of results. Note that these need to be stored in the index. If nothing is specified, defaults to document text and metadata fields.

Returns:

List of Documents most similar to the query.

similarity_search_with_score(query: str, k: int = 4, search_options: Dict[str, Any] | None = {}, **kwargs: Any) → List[Tuple[Document, float]]

Return documents that are most similar to the query with their scores.

Args:

query (str): Query to look up for similar documents k (int): Number of Documents to return.

Defaults to 4.

search_options (Optional[Dict[str, Any]]): Optional search options that are

passed to Couchbase search. Defaults to empty dictionary.

fields (Optional[List[str]]): Optional list of fields to include in the

metadata of results. Note that these need to be stored in the index. If nothing is specified, defaults to text and metadata fields.

Returns:

List of (Document, score) that are most similar to the query.

similarity_search_with_score_by_vector(embedding: List[float], k: int = 4, search_options: Dict[str, Any] | None = {}, **kwargs: Any) → List[Tuple[Document, float]]

Return docs most similar to embedding vector with their scores.

Args:

embedding (List[float]): Embedding vector to look up documents similar to. k (int): Number of Documents to return.

Defaults to 4.

search_options (Optional[Dict[str, Any]]): Optional search options that are

passed to Couchbase search. Defaults to empty dictionary.

fields (Optional[List[str]]): Optional list of fields to include in the

metadata of results. Note that these need to be stored in the index. If nothing is specified, defaults to all the fields stored in the index.

Returns:

List of (Document, score) that are the most similar to the query vector.

Module contents

class langchain_couchbase.CouchbaseCache(cluster: Cluster, bucket_name: str, scope_name: str, collection_name: str, ttl: timedelta | None = None, **kwargs: Dict[str, Any])

Bases: BaseCache

Couchbase LLM Cache LLM Cache that uses Couchbase as the backend

LLM = 'llm'

PROMPT = 'prompt'

RETURN_VAL = 'return_val'

clear(**kwargs: Any) → None

Clear the cache. This will delete all documents in the collection. This requires an index on the collection.

lookup(prompt: str, llm_string: str) → Sequence[Generation] | None

Look up from cache based on prompt and llm_string.

update(prompt: str, llm_string: str, return_val: Sequence[Generation]) → None

Update cache based on prompt and llm_string.

class langchain_couchbase.CouchbaseChatMessageHistory(*, cluster: Cluster, bucket_name: str, scope_name: str, collection_name: str, session_id: str, session_id_key: str = 'session_id', message_key: str = 'message', create_index: bool = True, ttl: timedelta | None = None)

Bases: BaseChatMessageHistory

Couchbase Chat Message History Chat message history that uses Couchbase as the storage

add_message(message: BaseMessage) → None

Add a message to the cache

add_messages(messages: Sequence[BaseMessage]) → None

Add messages to the cache in a batched manner

clear() → None

Clear the cache

property messages: List[BaseMessage]

Get all messages in the cache associated with the session_id

class langchain_couchbase.CouchbaseSemanticCache(cluster: Cluster, embedding: Embeddings, bucket_name: str, scope_name: str, collection_name: str, index_name: str, score_threshold: float | None = None, ttl: timedelta | None = None)

Bases: BaseCache, CouchbaseVectorStore

Couchbase Semantic Cache Cache backed by a Couchbase Server with Vector Store support

LLM = 'llm_string'

RETURN_VAL = 'return_val'

clear(**kwargs: Any) → None

Clear the cache. This will delete all documents in the collection. This requires an index on the collection.

lookup(prompt: str, llm_string: str) → Sequence[Generation] | None

Look up from cache based on the semantic similarity of the prompt

update(prompt: str, llm_string: str, return_val: Sequence[Generation]) → None

Update cache based on the prompt and llm_string

class langchain_couchbase.CouchbaseVectorStore(cluster: Cluster, bucket_name: str, scope_name: str, collection_name: str, embedding: Embeddings, index_name: str, *, text_key: str | None = 'text', embedding_key: str | None = 'embedding', scoped_index: bool = True)

Bases: VectorStore

__ModuleName__ vector store integration.

Setup:

Install langchain-couchbase and head over to the Couchbase [website](https://cloud.couchbase.com) and create a new connection, with a bucket, collection, and search index.

pip install -U langchain-couchbase

import getpass

COUCHBASE_CONNECTION_STRING = getpass.getpass("Enter the connection string for the Couchbase cluster: ")
DB_USERNAME = getpass.getpass("Enter the username for the Couchbase cluster: ")
DB_PASSWORD = getpass.getpass("Enter the password for the Couchbase cluster: ")

Key init args — indexing params:

embedding: Embeddings

Embedding function to use.

Key init args — client params:

cluster: Cluster

Couchbase cluster object with active connection.

bucket_name: str

Name of the bucket to store documents in.

scope_name: str

Name of the scope in the bucket to store documents in.

collection_name: str

Name of the collection in the scope to store documents in.

index_name: str

Name of the Search index to use.

Instantiate:

from datetime import timedelta
from langchain_openai import OpenAIEmbeddings
from couchbase.auth import PasswordAuthenticator
from couchbase.cluster import Cluster
from couchbase.options import ClusterOptions

auth = PasswordAuthenticator(DB_USERNAME, DB_PASSWORD)
options = ClusterOptions(auth)
cluster = Cluster(COUCHBASE_CONNECTION_STRING, options)

# Wait until the cluster is ready for use.
cluster.wait_until_ready(timedelta(seconds=5))

BUCKET_NAME = "langchain_bucket"
SCOPE_NAME = "_default"
COLLECTION_NAME = "default"
SEARCH_INDEX_NAME = "langchain-test-index"

vector_store = CouchbaseVectorStore(
    cluster=cluster,
    bucket_name=BUCKET_NAME,
    scope_name=SCOPE_NAME,
    collection_name=COLLECTION_NAME,
    embedding=embeddings,
    index_name=SEARCH_INDEX_NAME,
)

Add Documents:

from langchain_core.documents import Document

document_1 = Document(page_content="foo", metadata={"baz": "bar"})
document_2 = Document(page_content="thud", metadata={"bar": "baz"})
document_3 = Document(page_content="i will be deleted :(")

documents = [document_1, document_2, document_3]
ids = ["1", "2", "3"]
vector_store.add_documents(documents=documents, ids=ids)

Delete Documents:

vector_store.delete(ids=["3"])

# TODO: Fill out with example output. Search:

results = vector_store.similarity_search(query="thud",k=1)
for doc in results:
    print(f"* {doc.page_content} [{doc.metadata}]")

# TODO: Example output

# TODO: Fill out with relevant variables and example output. Search with filter:

# TODO: Update filter to correct format
results = vector_store.similarity_search(query="thud",k=1,filter={"bar": "baz"})
for doc in results:
    print(f"* {doc.page_content} [{doc.metadata}]")

# TODO: Example output

# TODO: Fill out with example output. Search with score:

results = vector_store.similarity_search_with_score(query="qux",k=1)
for doc, score in results:
    print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")

# TODO: Example output

# TODO: Fill out with example output. Async:

# add documents
# await vector_store.aadd_documents(documents=documents, ids=ids)

# delete documents
# await vector_store.adelete(ids=["3"])

# search
# results = vector_store.asimilarity_search(query="thud",k=1)

# search with score
results = await vector_store.asimilarity_search_with_score(query="qux",k=1)
for doc,score in results:
    print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")

# TODO: Example output

# TODO: Fill out with example output. Use as Retriever:

retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 1, "fetch_k": 2, "lambda_mult": 0.5},
)
retriever.invoke("thud")

# TODO: Example output

DEFAULT_BATCH_SIZE = 100

add_texts(texts: Iterable[str], metadatas: List[dict] | None = None, ids: List[str] | None = None, batch_size: int | None = None, **kwargs: Any) → List[str]

Run texts through the embeddings and persist in vectorstore.

If the document IDs are passed, the existing documents (if any) will be overwritten with the new ones.

Args:

texts (Iterable[str]): Iterable of strings to add to the vectorstore. metadatas (Optional[List[Dict]]): Optional list of metadatas associated

with the texts.

ids (Optional[List[str]]): Optional list of ids associated with the texts.

IDs have to be unique strings across the collection. If it is not specified uuids are generated and used as ids.

batch_size (Optional[int]): Optional batch size for bulk insertions.

Default is 100.

Returns:

List[str]:List of ids from adding the texts into the vectorstore.

delete(ids: List[str] | None = None, **kwargs: Any) → bool | None

Delete documents from the vector store by ids.

Args:

ids (List[str]): List of IDs of the documents to delete. batch_size (Optional[int]): Optional batch size for bulk deletions.

Returns:

bool: True if all the documents were deleted successfully, False otherwise.

property embeddings: Embeddings

Return the query embedding object.

classmethod from_texts(texts: List[str], embedding: Embeddings, metadatas: List[dict] | None = None, **kwargs: Any) → CouchbaseVectorStore

Construct a Couchbase vector store from a list of texts.

Example:

from langchain_couchbase import CouchbaseVectorStore from langchain_openai import OpenAIEmbeddings

from couchbase.cluster import Cluster from couchbase.auth import PasswordAuthenticator from couchbase.options import ClusterOptions from datetime import timedelta

auth = PasswordAuthenticator(username, password) options = ClusterOptions(auth) connect_string = “couchbases://localhost” cluster = Cluster(connect_string, options)

# Wait until the cluster is ready for use. cluster.wait_until_ready(timedelta(seconds=5))

embeddings = OpenAIEmbeddings()

texts = [“hello”, “world”]

vectorstore = CouchbaseVectorStore.from_texts(

texts, embedding=embeddings, cluster=cluster, bucket_name=””, scope_name=””, collection_name=””, index_name=”vector-index”,

)

Args:

texts (List[str]): list of texts to add to the vector store. embedding (Embeddings): embedding function to use. metadatas (optional[List[Dict]): list of metadatas to add to documents. **kwargs: Keyword arguments used to initialize the vector store with and/or

passed to add_texts method. Check the constructor and/or add_texts for the list of accepted arguments.

Returns:

A Couchbase vector store.

similarity_search(query: str, k: int = 4, search_options: Dict[str, Any] | None = {}, **kwargs: Any) → List[Document]

Return documents most similar to embedding vector with their scores.

Args:

query (str): Query to look up for similar documents k (int): Number of Documents to return.

Defaults to 4.

search_options (Optional[Dict[str, Any]]): Optional search options that are

passed to Couchbase search. Defaults to empty dictionary

fields (Optional[List[str]]): Optional list of fields to include in the

metadata of results. Note that these need to be stored in the index. If nothing is specified, defaults to all the fields stored in the index.

Returns:

List of Documents most similar to the query.

similarity_search_by_vector(embedding: List[float], k: int = 4, search_options: Dict[str, Any] | None = {}, **kwargs: Any) → List[Document]

Return documents that are most similar to the vector embedding.

Args:

embedding (List[float]): Embedding to look up documents similar to. k (int): Number of Documents to return.

Defaults to 4.

search_options (Optional[Dict[str, Any]]): Optional search options that are

passed to Couchbase search. Defaults to empty dictionary.

fields (Optional[List[str]]): Optional list of fields to include in the

metadata of results. Note that these need to be stored in the index. If nothing is specified, defaults to document text and metadata fields.

Returns:

List of Documents most similar to the query.

similarity_search_with_score(query: str, k: int = 4, search_options: Dict[str, Any] | None = {}, **kwargs: Any) → List[Tuple[Document, float]]

Return documents that are most similar to the query with their scores.

Args:

query (str): Query to look up for similar documents k (int): Number of Documents to return.

Defaults to 4.

search_options (Optional[Dict[str, Any]]): Optional search options that are

passed to Couchbase search. Defaults to empty dictionary.

fields (Optional[List[str]]): Optional list of fields to include in the

metadata of results. Note that these need to be stored in the index. If nothing is specified, defaults to text and metadata fields.

Returns:

List of (Document, score) that are most similar to the query.

similarity_search_with_score_by_vector(embedding: List[float], k: int = 4, search_options: Dict[str, Any] | None = {}, **kwargs: Any) → List[Tuple[Document, float]]

Return docs most similar to embedding vector with their scores.

Args:

embedding (List[float]): Embedding vector to look up documents similar to. k (int): Number of Documents to return.

Defaults to 4.

search_options (Optional[Dict[str, Any]]): Optional search options that are

passed to Couchbase search. Defaults to empty dictionary.

fields (Optional[List[str]]): Optional list of fields to include in the

metadata of results. Note that these need to be stored in the index. If nothing is specified, defaults to all the fields stored in the index.

Returns:

List of (Document, score) that are the most similar to the query vector.