From fa7707c1488c75462634b905a7277e9d26858065 Mon Sep 17 00:00:00 2001 From: Rodrigo Date: Mon, 17 Jun 2024 11:36:32 -0300 Subject: [PATCH] refactor vector stores --- .../components/vectorstores/AstraDB.py | 406 +++++++++++------- .../components/vectorstores/Cassandra.py | 181 ++++---- .../components/vectorstores/Chroma.py | 280 ++++++++---- .../components/vectorstores/Couchbase.py | 153 ++++--- .../langflow/components/vectorstores/FAISS.py | 212 +++++++-- .../vectorstores/MongoDBAtlasVector.py | 121 ++++-- .../components/vectorstores/Pinecone.py | 208 ++++----- .../components/vectorstores/Qdrant.py | 201 ++++----- .../vectorstores/SupabaseVectorStore.py | 106 +++-- .../components/vectorstores/Upstash.py | 162 +++---- .../components/vectorstores/Vectara.py | 144 +++---- .../components/vectorstores/Weaviate.py | 173 ++++---- .../components/vectorstores/pgvector.py | 141 +++--- 13 files changed, 1442 insertions(+), 1046 deletions(-) diff --git a/src/backend/base/langflow/components/vectorstores/AstraDB.py b/src/backend/base/langflow/components/vectorstores/AstraDB.py index e912557c2..3d40c5258 100644 --- a/src/backend/base/langflow/components/vectorstores/AstraDB.py +++ b/src/backend/base/langflow/components/vectorstores/AstraDB.py @@ -1,115 +1,169 @@ -from typing import List, Optional, Union - -from langchain_core.retrievers import BaseRetriever - -from langflow.custom import CustomComponent -from langflow.field_typing import Embeddings, VectorStore +from langflow.custom import Component +from langflow.field_typing import Text +from langflow.inputs import ( + StrInput, + IntInput, + BoolInput, + DropdownInput, + MultilineInput, + HandleInput, +) from langflow.schema import Data +from langflow.template import Output +from langflow.field_typing import Embeddings +from loguru import logger -class AstraDBVectorStoreComponent(CustomComponent): - display_name = "Astra DB" - description = "Builds or loads an Astra DB Vector Store." - icon = "AstraDB" - field_order = ["token", "api_endpoint", "collection_name", "inputs", "embedding"] +class AstraDBComponent(Component): + display_name: str = "Astra DB Vector Store" + description: str = "Implementation of Vector Store using Astra DB with search capabilities" + documentation: str = "https://python.langchain.com/docs/integrations/vectorstores/astradb" + icon: str = "AstraDB" - def build_config(self): - return { - "inputs": { - "display_name": "Inputs", - "info": "Optional list of data to be processed and stored in the vector store.", - }, - "embedding": {"display_name": "Embedding", "info": "Embedding to use"}, - "collection_name": { - "display_name": "Collection Name", - "info": "The name of the collection within Astra DB where the vectors will be stored.", - }, - "token": { - "display_name": "Astra DB Application Token", - "info": "Authentication token for accessing Astra DB.", - "password": True, - }, - "api_endpoint": { - "display_name": "API Endpoint", - "info": "API endpoint URL for the Astra DB service.", - }, - "namespace": { - "display_name": "Namespace", - "info": "Optional namespace within Astra DB to use for the collection.", - "advanced": True, - }, - "metric": { - "display_name": "Metric", - "info": "Optional distance metric for vector comparisons in the vector store.", - "advanced": True, - }, - "batch_size": { - "display_name": "Batch Size", - "info": "Optional number of data to process in a single batch.", - "advanced": True, - }, - "bulk_insert_batch_concurrency": { - "display_name": "Bulk Insert Batch Concurrency", - "info": "Optional concurrency level for bulk insert operations.", - "advanced": True, - }, - "bulk_insert_overwrite_concurrency": { - "display_name": "Bulk Insert Overwrite Concurrency", - "info": "Optional concurrency level for bulk insert operations that overwrite existing data.", - "advanced": True, - }, - "bulk_delete_concurrency": { - "display_name": "Bulk Delete Concurrency", - "info": "Optional concurrency level for bulk delete operations.", - "advanced": True, - }, - "setup_mode": { - "display_name": "Setup Mode", - "info": "Configuration mode for setting up the vector store, with options like “Sync”, “Async”, or “Off”.", - "options": ["Sync", "Async", "Off"], - "advanced": True, - }, - "pre_delete_collection": { - "display_name": "Pre Delete Collection", - "info": "Boolean flag to determine whether to delete the collection before creating a new one.", - "advanced": True, - }, - "metadata_indexing_include": { - "display_name": "Metadata Indexing Include", - "info": "Optional list of metadata fields to include in the indexing.", - "advanced": True, - }, - "metadata_indexing_exclude": { - "display_name": "Metadata Indexing Exclude", - "info": "Optional list of metadata fields to exclude from the indexing.", - "advanced": True, - }, - "collection_indexing_policy": { - "display_name": "Collection Indexing Policy", - "info": "Optional dictionary defining the indexing policy for the collection.", - "advanced": True, - }, - } + inputs = [ + StrInput( + name="collection_name", + display_name="Collection Name", + info="The name of the collection within Astra DB where the vectors will be stored.", + ), + StrInput( + name="token", + display_name="Astra DB Application Token", + info="Authentication token for accessing Astra DB.", + password=True, + ), + StrInput( + name="api_endpoint", + display_name="API Endpoint", + info="API endpoint URL for the Astra DB service.", + ), + StrInput( + name="code", + display_name="Code", + advanced=True, + ), + HandleInput( + name="vector_store_inputs", + display_name="Vector Store Inputs", + input_types=["Document", "Data"], + is_list=True, + ), + HandleInput( + name="embedding", + display_name="Embedding", + input_types=["Embeddings"], + ), + StrInput( + name="namespace", + display_name="Namespace", + info="Optional namespace within Astra DB to use for the collection.", + advanced=True, + ), + DropdownInput( + name="metric", + display_name="Metric", + info="Optional distance metric for vector comparisons in the vector store.", + options=["cosine", "dot_product", "euclidean"], + advanced=True, + ), + IntInput( + name="batch_size", + display_name="Batch Size", + info="Optional number of data to process in a single batch.", + advanced=True, + ), + IntInput( + name="bulk_insert_batch_concurrency", + display_name="Bulk Insert Batch Concurrency", + info="Optional concurrency level for bulk insert operations.", + advanced=True, + ), + IntInput( + name="bulk_insert_overwrite_concurrency", + display_name="Bulk Insert Overwrite Concurrency", + info="Optional concurrency level for bulk insert operations that overwrite existing data.", + advanced=True, + ), + IntInput( + name="bulk_delete_concurrency", + display_name="Bulk Delete Concurrency", + info="Optional concurrency level for bulk delete operations.", + advanced=True, + ), + DropdownInput( + name="setup_mode", + display_name="Setup Mode", + info="Configuration mode for setting up the vector store, with options like 'Sync', 'Async', or 'Off'.", + options=["Sync", "Async", "Off"], + advanced=True, + ), + BoolInput( + name="pre_delete_collection", + display_name="Pre Delete Collection", + info="Boolean flag to determine whether to delete the collection before creating a new one.", + advanced=True, + ), + StrInput( + name="metadata_indexing_include", + display_name="Metadata Indexing Include", + info="Optional list of metadata fields to include in the indexing.", + advanced=True, + ), + StrInput( + name="metadata_indexing_exclude", + display_name="Metadata Indexing Exclude", + info="Optional list of metadata fields to exclude from the indexing.", + advanced=True, + ), + StrInput( + name="collection_indexing_policy", + display_name="Collection Indexing Policy", + info="Optional dictionary defining the indexing policy for the collection.", + advanced=True, + ), + BoolInput( + name="add_to_vector_store", + display_name="Add to Vector Store", + info="If true, the Vector Store Inputs will be added to the Vector Store.", + ), + MultilineInput( + name="search_input", + display_name="Search Input", + ), + DropdownInput( + name="search_type", + display_name="Search Type", + options=["Similarity", "MMR"], + value="Similarity", + ), + IntInput( + name="number_of_results", + display_name="Number of Results", + info="Number of results to return.", + advanced=True, + value=4, + ), + ] - def build( - self, - embedding: Embeddings, - token: str, - api_endpoint: str, - collection_name: str, - inputs: Optional[List[Data]] = None, - namespace: Optional[str] = None, - metric: Optional[str] = None, - batch_size: Optional[int] = None, - bulk_insert_batch_concurrency: Optional[int] = None, - bulk_insert_overwrite_concurrency: Optional[int] = None, - bulk_delete_concurrency: Optional[int] = None, - setup_mode: str = "Sync", - pre_delete_collection: bool = False, - metadata_indexing_include: Optional[List[str]] = None, - metadata_indexing_exclude: Optional[List[str]] = None, - collection_indexing_policy: Optional[dict] = None, - ) -> Union[VectorStore, BaseRetriever]: + outputs = [ + Output( + display_name="Vector Store", + name="vector_store", + method="build_vector_store", + ), + Output( + display_name="Base Retriever", + name="base_retriever", + method="build_base_retriever", + ), + Output( + display_name="Search Results", + name="search_results", + method="search_documents", + ), + ] + + def build_vector_store(self): try: from langchain_astradb import AstraDBVectorStore from langchain_astradb.utils.astradb import SetupMode @@ -120,47 +174,105 @@ class AstraDBVectorStoreComponent(CustomComponent): ) try: - setup_mode_value = SetupMode[setup_mode.upper()] + setup_mode_value = SetupMode[self.setup_mode.upper()] except KeyError: - raise ValueError(f"Invalid setup mode: {setup_mode}") - if inputs: - documents = [_input.to_lc_document() for _input in inputs] + raise ValueError(f"Invalid setup mode: {self.setup_mode}") - vector_store = AstraDBVectorStore.from_documents( - documents=documents, - embedding=embedding, - collection_name=collection_name, - token=token, - api_endpoint=api_endpoint, - namespace=namespace, - metric=metric, - batch_size=batch_size, - bulk_insert_batch_concurrency=bulk_insert_batch_concurrency, - bulk_insert_overwrite_concurrency=bulk_insert_overwrite_concurrency, - bulk_delete_concurrency=bulk_delete_concurrency, - setup_mode=setup_mode_value, - pre_delete_collection=pre_delete_collection, - metadata_indexing_include=metadata_indexing_include, - metadata_indexing_exclude=metadata_indexing_exclude, - collection_indexing_policy=collection_indexing_policy, - ) - else: - vector_store = AstraDBVectorStore( - embedding=embedding, - collection_name=collection_name, - token=token, - api_endpoint=api_endpoint, - namespace=namespace, - metric=metric, - batch_size=batch_size, - bulk_insert_batch_concurrency=bulk_insert_batch_concurrency, - bulk_insert_overwrite_concurrency=bulk_insert_overwrite_concurrency, - bulk_delete_concurrency=bulk_delete_concurrency, - setup_mode=setup_mode_value, - pre_delete_collection=pre_delete_collection, - metadata_indexing_include=metadata_indexing_include, - metadata_indexing_exclude=metadata_indexing_exclude, - collection_indexing_policy=collection_indexing_policy, - ) + vector_store_kwargs = { + "embedding": self.embedding, + "collection_name": self.collection_name, + "token": self.token, + "api_endpoint": self.api_endpoint, + "namespace": self.namespace, + "metric": self.metric, + "batch_size": self.batch_size, + "bulk_insert_batch_concurrency": self.bulk_insert_batch_concurrency, + "bulk_insert_overwrite_concurrency": self.bulk_insert_overwrite_concurrency, + "bulk_delete_concurrency": self.bulk_delete_concurrency, + "setup_mode": setup_mode_value, + "pre_delete_collection": self.pre_delete_collection, + } + if self.metadata_indexing_include: + vector_store_kwargs["metadata_indexing_include"] = self.metadata_indexing_include + elif self.metadata_indexing_exclude: + vector_store_kwargs["metadata_indexing_exclude"] = self.metadata_indexing_exclude + elif self.collection_indexing_policy: + vector_store_kwargs["collection_indexing_policy"] = self.collection_indexing_policy + + try: + vector_store = AstraDBVectorStore(**vector_store_kwargs) + except Exception as e: + raise ValueError(f"Error initializing AstraDBVectorStore: {str(e)}") from e + + if self.add_to_vector_store: + self._add_documents_to_vector_store(vector_store) + + self.status = self._astradb_collection_to_data(vector_store.collection) return vector_store + + def build_base_retriever(self): + return self.build_vector_store() + + def _add_documents_to_vector_store(self, vector_store): + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + raise ValueError("Vector Store Inputs must be Data objects.") + + if documents and self.embedding is not None: + logger.debug(f"Adding {len(documents)} documents to the Vector Store.") + try: + vector_store.add_documents(documents) + except Exception as e: + raise ValueError(f"Error adding documents to AstraDBVectorStore: {str(e)}") from e + else: + logger.debug("No documents to add to the Vector Store.") + + def search_documents(self): + vector_store = self.build_vector_store() + + logger.debug(f"Search input: {self.search_input}") + logger.debug(f"Search type: {self.search_type}") + logger.debug(f"Number of results: {self.number_of_results}") + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + try: + if self.search_type == "Similarity": + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + elif self.search_type == "MMR": + docs = vector_store.max_marginal_relevance_search( + query=self.search_input, + k=self.number_of_results, + ) + else: + raise ValueError(f"Invalid search type: {self.search_type}") + except Exception as e: + raise ValueError(f"Error performing search in AstraDBVectorStore: {str(e)}") from e + + logger.debug(f"Retrieved documents: {len(docs)}") + + data = self._docs_to_data(docs) + logger.debug(f"Converted documents to data: {len(data)}") + self.status = data + return data + else: + logger.debug("No search input provided. Skipping search.") + return [] + + def _astradb_collection_to_data(self, collection): + data = [] + for item in collection["data"]: + data.append(Data(content=item["content"])) + return data + + def _docs_to_data(self, docs): + data = [] + for doc in docs: + data.append(Data(content=doc.page_content)) + return data diff --git a/src/backend/base/langflow/components/vectorstores/Cassandra.py b/src/backend/base/langflow/components/vectorstores/Cassandra.py index b5fb76dc1..a68e6a4bf 100644 --- a/src/backend/base/langflow/components/vectorstores/Cassandra.py +++ b/src/backend/base/langflow/components/vectorstores/Cassandra.py @@ -2,79 +2,51 @@ from typing import Any, List, Optional, Tuple from langchain_community.utilities.cassandra import SetupMode from langchain_community.vectorstores import Cassandra +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever -from langflow.custom import CustomComponent -from langflow.field_typing import Embeddings, VectorStore +from langflow.custom import Component +from langflow.field_typing import Embeddings, Text from langflow.schema import Data +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput, DropdownInput +from langflow.template import Output +from langflow.helpers.data import docs_to_data - -class CassandraVectorStoreComponent(CustomComponent): +class CassandraVectorStoreComponent(Component): display_name = "Cassandra" - description = "Builds or loads a Cassandra Vector Store." + description = "Cassandra Vector Store with search capabilities" + documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/cassandra" icon = "Cassandra" - field_order = ["token", "database_id", "table_name", "inputs", "embedding"] - def build_config(self): - return { - "inputs": { - "display_name": "Inputs", - "info": "Optional list of data to be processed and stored in the vector store.", - }, - "embedding": {"display_name": "Embedding", "info": "Embedding to use"}, - "token": { - "display_name": "Token", - "info": "Authentication token for accessing Cassandra on Astra DB.", - "password": True, - }, - "database_id": { - "display_name": "Database ID", - "info": "The Astra database ID.", - }, - "table_name": { - "display_name": "Table Name", - "info": "The name of the table where vectors will be stored.", - }, - "keyspace": { - "display_name": "Keyspace", - "info": "Optional key space within Astra DB. The keyspace should already be created.", - "advanced": True, - }, - "ttl_seconds": { - "display_name": "TTL Seconds", - "info": "Optional time-to-live for the added texts.", - "advanced": True, - }, - "batch_size": { - "display_name": "Batch Size", - "info": "Optional number of data to process in a single batch.", - "advanced": True, - }, - "body_index_options": { - "display_name": "Body Index Options", - "info": "Optional options used to create the body index.", - "advanced": True, - }, - "setup_mode": { - "display_name": "Setup Mode", - "info": "Configuration mode for setting up the Cassandra table, with options like 'Sync', 'Async', or 'Off'.", - "options": ["Sync", "Async", "Off"], - "advanced": True, - }, - } + inputs = [ + StrInput(name="token", display_name="Token", info="Authentication token for accessing Cassandra on Astra DB.", password=True, required=True), + StrInput(name="database_id", display_name="Database ID", info="The Astra database ID.", required=True), + StrInput(name="table_name", display_name="Table Name", info="The name of the table where vectors will be stored.", required=True), + StrInput(name="keyspace", display_name="Keyspace", info="Optional key space within Astra DB. The keyspace should already be created.", advanced=True), + IntInput(name="ttl_seconds", display_name="TTL Seconds", info="Optional time-to-live for the added texts.", advanced=True), + IntInput(name="batch_size", display_name="Batch Size", info="Optional number of data to process in a single batch.", value=16, advanced=True), + StrInput(name="body_index_options", display_name="Body Index Options", info="Optional options used to create the body index.", advanced=True), + DropdownInput(name="setup_mode", display_name="Setup Mode", info="Configuration mode for setting up the Cassandra table, with options like 'Sync', 'Async', or 'Off'.", options=["Sync", "Async", "Off"], value="Sync", advanced=True), + HandleInput(name="embedding", display_name="Embedding", input_types=["Embeddings"]), + HandleInput(name="vector_store_inputs", display_name="Vector Store Inputs", input_types=["Document", "Data"], is_list=True), + BoolInput(name="add_to_vector_store", display_name="Add to Vector Store", info="If true, the Vector Store Inputs will be added to the Vector Store."), + StrInput(name="search_input", display_name="Search Input"), + IntInput(name="number_of_results", display_name="Number of Results", info="Number of results to return.", value=4, advanced=True), + ] - def build( - self, - embedding: Embeddings, - token: str, - database_id: str, - inputs: Optional[List[Data]] = None, - keyspace: Optional[str] = None, - table_name: str = "", - ttl_seconds: Optional[int] = None, - batch_size: int = 16, - body_index_options: Optional[List[Tuple[str, Any]]] = None, - setup_mode: SetupMode = SetupMode.SYNC, - ) -> VectorStore: + outputs = [ + Output(display_name="Vector Store", name="vector_store", method="build_vector_store", output_type=Cassandra), + Output(display_name="Base Retriever", name="base_retriever", method="build_base_retriever", output_type=BaseRetriever), + Output(display_name="Search Results", name="search_results", method="search_documents"), + ] + + def build_vector_store(self) -> Cassandra: + return self._build_cassandra() + + def build_base_retriever(self) -> BaseRetriever: + return self._build_cassandra() + + def _build_cassandra(self) -> Cassandra: try: import cassio except ImportError: @@ -83,29 +55,68 @@ class CassandraVectorStoreComponent(CustomComponent): ) cassio.init( - database_id=database_id, - token=token, + database_id=self.database_id, + token=self.token, ) - if inputs: - documents = [_input.to_lc_document() for _input in inputs] - table = Cassandra.from_documents( - documents=documents, - embedding=embedding, - table_name=table_name, - keyspace=keyspace, - ttl_seconds=ttl_seconds, - batch_size=batch_size, - body_index_options=body_index_options, - ) + if self.add_to_vector_store: + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + documents.append(_input) + + if documents: + table = Cassandra.from_documents( + documents=documents, + embedding=self.embedding, + table_name=self.table_name, + keyspace=self.keyspace, + ttl_seconds=self.ttl_seconds, + batch_size=self.batch_size, + body_index_options=self.body_index_options, + ) + else: + table = Cassandra( + embedding=self.embedding, + table_name=self.table_name, + keyspace=self.keyspace, + ttl_seconds=self.ttl_seconds, + body_index_options=self.body_index_options, + setup_mode=self.setup_mode, + ) else: table = Cassandra( - embedding=embedding, - table_name=table_name, - keyspace=keyspace, - ttl_seconds=ttl_seconds, - body_index_options=body_index_options, - setup_mode=setup_mode, + embedding=self.embedding, + table_name=self.table_name, + keyspace=self.keyspace, + ttl_seconds=self.ttl_seconds, + body_index_options=self.body_index_options, + setup_mode=self.setup_mode, ) return table + + def search_documents(self) -> List[Data]: + vector_store = self._build_cassandra() + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + try: + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + except KeyError as e: + if "content" in str(e): + raise ValueError( + "You should ingest data through Langflow (or LangChain) to query it in Langflow. Your collection does not contain a field name 'content'." + ) + else: + raise e + + data = docs_to_data(docs) + self.status = data + return data + else: + return [] diff --git a/src/backend/base/langflow/components/vectorstores/Chroma.py b/src/backend/base/langflow/components/vectorstores/Chroma.py index 8aac051ca..62735d2f5 100644 --- a/src/backend/base/langflow/components/vectorstores/Chroma.py +++ b/src/backend/base/langflow/components/vectorstores/Chroma.py @@ -3,116 +3,182 @@ from typing import List, Optional, Union import chromadb from chromadb.config import Settings -from langchain_chroma import Chroma -from langchain_core.embeddings import Embeddings -from langchain_core.retrievers import BaseRetriever -from langchain_core.vectorstores import VectorStore +from langchain.vectorstores import Chroma +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever from langflow.base.vectorstores.utils import chroma_collection_to_data -from langflow.custom import CustomComponent +from langflow.custom import Component +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput, DropdownInput from langflow.schema import Data +from langflow.template import Output +from langflow.field_typing import Embeddings +from langflow.helpers.data import docs_to_data +from loguru import logger -class ChromaComponent(CustomComponent): +class ChromaVectorStoreComponent(Component): """ - A custom component for implementing a Vector Store using Chroma. + Chroma Vector Store with search capabilities """ - display_name: str = "Chroma" - description: str = "Implementation of Vector Store using Chroma" + display_name: str = "Chroma DB" + description: str = "Chroma Vector Store with search capabilities" documentation = "https://python.langchain.com/docs/integrations/vectorstores/chroma" icon = "Chroma" - def build_config(self): + inputs = [ + StrInput( + name="collection_name", + display_name="Collection Name", + value="langflow", + ), + StrInput( + name="persist_directory", + display_name="Persist Directory", + ), + StrInput( + name="code", + display_name="Code", + advanced=True, + ), + StrInput( + name="vector_store_inputs", + display_name="Vector Store Inputs", + input_types=["Document", "Data"], + is_list=True + ), + HandleInput( + name="embedding", + display_name="Embedding", + input_types=["Embeddings"] + ), + StrInput( + name="chroma_server_cors_allow_origins", + display_name="Server CORS Allow Origins", + advanced=True, + ), + StrInput( + name="chroma_server_host", + display_name="Server Host", + advanced=True, + ), + IntInput( + name="chroma_server_http_port", + display_name="Server HTTP Port", + advanced=True, + ), + IntInput( + name="chroma_server_grpc_port", + display_name="Server gRPC Port", + advanced=True, + ), + BoolInput( + name="chroma_server_ssl_enabled", + display_name="Server SSL Enabled", + advanced=True, + ), + BoolInput( + name="allow_duplicates", + display_name="Allow Duplicates", + advanced=True, + info="If false, will not add documents that are already in the Vector Store.", + ), + BoolInput( + name="add_to_vector_store", + display_name="Add to Vector Store", + info="If true, the Vector Store Inputs will be added to the Vector Store.", + ), + StrInput( + name="search_input", + display_name="Search Input", + ), + DropdownInput( + name="search_type", + display_name="Search Type", + options=["Similarity", "MMR"], + value="Similarity", + ), + IntInput( + name="number_of_results", + display_name="Number of Results", + info="Number of results to return.", + advanced=True, + value=4, + ), + ] + + outputs = [ + Output( + display_name="Vector Store", + name="vector_store", + method="build_vector_store", + ), + Output( + display_name="Base Retriever", + name="base_retriever", + method="build_base_retriever", + ), + Output( + display_name="Search Results", + name="search_results", + method="search_documents", + ), + ] + + def build_vector_store(self) -> Chroma: """ - Builds the configuration for the component. - - Returns: - - dict: A dictionary containing the configuration options for the component. + Builds the Vector Store object. """ - return { - "collection_name": {"display_name": "Collection Name", "value": "langflow"}, - "index_directory": {"display_name": "Persist Directory"}, - "code": {"advanced": True, "display_name": "Code"}, - "inputs": {"display_name": "Input", "input_types": ["Document", "Data"]}, - "embedding": {"display_name": "Embedding"}, - "chroma_server_cors_allow_origins": { - "display_name": "Server CORS Allow Origins", - "advanced": True, - }, - "chroma_server_host": {"display_name": "Server Host", "advanced": True}, - "chroma_server_http_port": {"display_name": "Server HTTP Port", "advanced": True}, - "chroma_server_grpc_port": { - "display_name": "Server gRPC Port", - "advanced": True, - }, - "chroma_server_ssl_enabled": { - "display_name": "Server SSL Enabled", - "advanced": True, - }, - "allow_duplicates": { - "display_name": "Allow Duplicates", - "advanced": True, - "info": "If false, will not add documents that are already in the Vector Store.", - }, - } + return self._build_chroma() - def build( - self, - collection_name: str, - embedding: Embeddings, - chroma_server_ssl_enabled: bool, - index_directory: Optional[str] = None, - inputs: Optional[List[Data]] = None, - chroma_server_cors_allow_origins: List[str] = [], - chroma_server_host: Optional[str] = None, - chroma_server_http_port: Optional[int] = None, - chroma_server_grpc_port: Optional[int] = None, - allow_duplicates: bool = False, - ) -> Union[VectorStore, BaseRetriever]: + def build_base_retriever(self) -> BaseRetriever: """ - Builds the Vector Store or BaseRetriever object. - - Args: - - collection_name (str): The name of the collection. - - embedding (Embeddings): The embeddings to use for the Vector Store. - - chroma_server_ssl_enabled (bool): Whether to enable SSL for the Chroma server. - - index_directory (Optional[str]): The directory to persist the Vector Store to. - - inputs (Optional[List[Data]]): The input data to use for the Vector Store. - - chroma_server_cors_allow_origins (List[str]): The CORS allow origins for the Chroma server. - - chroma_server_host (Optional[str]): The host for the Chroma server. - - chroma_server_http_port (Optional[int]): The HTTP port for the Chroma server. - - chroma_server_grpc_port (Optional[int]): The gRPC port for the Chroma server. - - allow_duplicates (bool): Whether to allow duplicates in the Vector Store. - - Returns: - - Union[VectorStore, BaseRetriever]: The Vector Store or BaseRetriever object. + Builds the BaseRetriever object. """ + return self._build_chroma() + def _build_chroma(self) -> Chroma: + """ + Builds the Chroma object. + """ # Chroma settings chroma_settings = None client = None - if chroma_server_host is not None: + if self.chroma_server_host: chroma_settings = Settings( - chroma_server_cors_allow_origins=chroma_server_cors_allow_origins or [], - chroma_server_host=chroma_server_host, - chroma_server_http_port=chroma_server_http_port or None, - chroma_server_grpc_port=chroma_server_grpc_port or None, - chroma_server_ssl_enabled=chroma_server_ssl_enabled, + chroma_server_cors_allow_origins=self.chroma_server_cors_allow_origins or [], + chroma_server_host=self.chroma_server_host, + chroma_server_http_port=self.chroma_server_http_port or None, + chroma_server_grpc_port=self.chroma_server_grpc_port or None, + chroma_server_ssl_enabled=self.chroma_server_ssl_enabled, ) - client = chromadb.HttpClient(settings=chroma_settings) + client = chromadb.Client(settings=chroma_settings) - # Check index_directory and expand it if it is a relative path - if index_directory is not None: - index_directory = self.resolve_path(index_directory) + # Check persist_directory and expand it if it is a relative path + if self.persist_directory is not None: + persist_directory = self.resolve_path(self.persist_directory) + else: + persist_directory = None chroma = Chroma( - persist_directory=index_directory, + persist_directory=persist_directory, client=client, - embedding_function=embedding, - collection_name=collection_name, + embedding_function=self.embedding, + collection_name=self.collection_name, ) - if allow_duplicates: + + if self.add_to_vector_store: + self._add_documents_to_vector_store(chroma) + + self.status = chroma_collection_to_data(chroma.get()) + return chroma + + def _add_documents_to_vector_store(self, chroma: Chroma) -> None: + """ + Adds documents to the Vector Store. + """ + if self.allow_duplicates: stored_data = [] else: stored_data = chroma_collection_to_data(chroma.get()) @@ -120,16 +186,54 @@ class ChromaComponent(CustomComponent): for value in deepcopy(stored_data): del value.id _stored_documents_without_id.append(value) + documents = [] - for _input in inputs or []: + for _input in self.vector_store_inputs or []: if isinstance(_input, Data): if _input not in _stored_documents_without_id: documents.append(_input.to_lc_document()) else: - raise ValueError("Inputs must be a Data objects.") + raise ValueError("Vector Store Inputs must be Data objects.") - if documents and embedding is not None: + if documents and self.embedding is not None: + logger.debug(f"Adding {len(documents)} documents to the Vector Store.") chroma.add_documents(documents) + else: + logger.debug("No documents to add to the Vector Store.") - self.status = stored_data - return chroma + def search_documents(self) -> List[Data]: + """ + Search for documents in the Chroma vector store. + """ + if not self.search_input: + return + + vector_store = self._build_chroma() + + logger.debug(f"Search input: {self.search_input}") + logger.debug(f"Search type: {self.search_type}") + logger.debug(f"Number of results: {self.number_of_results}") + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + if self.search_type == "Similarity": + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + elif self.search_type == "MMR": + docs = vector_store.max_marginal_relevance_search( + query=self.search_input, + k=self.number_of_results, + ) + else: + raise ValueError(f"Invalid search type: {self.search_type}") + + logger.debug(f"Retrieved documents: {len(docs)}") + + data = docs_to_data(docs) + logger.debug(f"Converted documents to data: {len(data)}") + self.status = data + return data + else: + logger.debug("No search input provided. Skipping search.") + return [] diff --git a/src/backend/base/langflow/components/vectorstores/Couchbase.py b/src/backend/base/langflow/components/vectorstores/Couchbase.py index fe09a3f3b..f79d642b0 100644 --- a/src/backend/base/langflow/components/vectorstores/Couchbase.py +++ b/src/backend/base/langflow/components/vectorstores/Couchbase.py @@ -1,94 +1,119 @@ from datetime import timedelta from typing import List, Optional, Union -from langchain_core.retrievers import BaseRetriever +from langchain_community.vectorstores import CouchbaseVectorStore +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever -from langflow.custom import CustomComponent -from langflow.field_typing import Embeddings, VectorStore +from langflow.custom import Component +from langflow.field_typing import Embeddings, Text from langflow.schema import Data +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput +from langflow.template import Output +from langflow.helpers.data import docs_to_data - -class CouchbaseComponent(CustomComponent): +class CouchbaseVectorStoreComponent(Component): display_name = "Couchbase" - description = "Construct a `Couchbase Vector Search` vector store from raw documents." - documentation = "https://python.langchain.com/docs/integrations/vectorstores/couchbase" + description = "Couchbase Vector Store with search capabilities" + documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/couchbase" icon = "Couchbase" - field_order = [ - "couchbase_connection_string", - "couchbase_username", - "couchbase_password", - "bucket_name", - "scope_name", - "collection_name", - "index_name", + + inputs = [ + StrInput(name="couchbase_connection_string", display_name="Couchbase Cluster connection string", required=True), + StrInput(name="couchbase_username", display_name="Couchbase username", required=True), + StrInput(name="couchbase_password", display_name="Couchbase password", password=True, required=True), + StrInput(name="bucket_name", display_name="Bucket Name", required=True), + StrInput(name="scope_name", display_name="Scope Name", required=True), + StrInput(name="collection_name", display_name="Collection Name", required=True), + StrInput(name="index_name", display_name="Index Name", required=True), + HandleInput(name="embedding", display_name="Embedding", input_types=["Embeddings"]), + HandleInput(name="vector_store_inputs", display_name="Vector Store Inputs", input_types=["Document", "Data"], is_list=True), + BoolInput(name="add_to_vector_store", display_name="Add to Vector Store", info="If true, the Vector Store Inputs will be added to the Vector Store."), + StrInput(name="search_input", display_name="Search Input"), + IntInput(name="number_of_results", display_name="Number of Results", info="Number of results to return.", value=4, advanced=True), ] - def build_config(self): - return { - "inputs": {"display_name": "Input", "input_types": ["Document", "Data"]}, - "embedding": {"display_name": "Embedding"}, - "couchbase_connection_string": {"display_name": "Couchbase Cluster connection string", "required": True}, - "couchbase_username": {"display_name": "Couchbase username", "required": True}, - "couchbase_password": {"display_name": "Couchbase password", "password": True, "required": True}, - "bucket_name": {"display_name": "Bucket Name", "required": True}, - "scope_name": {"display_name": "Scope Name", "required": True}, - "collection_name": {"display_name": "Collection Name", "required": True}, - "index_name": {"display_name": "Index Name", "required": True}, - } + outputs = [ + Output(display_name="Vector Store", name="vector_store", method="build_vector_store", output_type=CouchbaseVectorStore), + Output(display_name="Base Retriever", name="base_retriever", method="build_base_retriever", output_type=BaseRetriever), + Output(display_name="Search Results", name="search_results", method="search_documents"), + ] - def build( - self, - embedding: Embeddings, - inputs: Optional[List[Data]] = None, - bucket_name: str = "", - scope_name: str = "", - collection_name: str = "", - index_name: str = "", - couchbase_connection_string: str = "", - couchbase_username: str = "", - couchbase_password: str = "", - ) -> Union[VectorStore, BaseRetriever]: + def build_vector_store(self) -> CouchbaseVectorStore: + return self._build_couchbase() + + def build_base_retriever(self) -> BaseRetriever: + return self._build_couchbase() + + def _build_couchbase(self) -> CouchbaseVectorStore: try: from couchbase.auth import PasswordAuthenticator # type: ignore from couchbase.cluster import Cluster # type: ignore from couchbase.options import ClusterOptions # type: ignore - from langchain_community.vectorstores import CouchbaseVectorStore except ImportError as e: raise ImportError( "Failed to import Couchbase dependencies. Install it using `pip install langflow[couchbase] --pre`" ) from e try: - auth = PasswordAuthenticator(couchbase_username, couchbase_password) + auth = PasswordAuthenticator(self.couchbase_username, self.couchbase_password) options = ClusterOptions(auth) - cluster = Cluster(couchbase_connection_string, options) + cluster = Cluster(self.couchbase_connection_string, options) cluster.wait_until_ready(timedelta(seconds=5)) except Exception as e: raise ValueError(f"Failed to connect to Couchbase: {e}") - documents = [] - for _input in inputs or []: - if isinstance(_input, Data): - documents.append(_input.to_lc_document()) + + if self.add_to_vector_store: + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + documents.append(_input) + + if documents: + couchbase_vs = CouchbaseVectorStore.from_documents( + documents=documents, + cluster=cluster, + bucket_name=self.bucket_name, + scope_name=self.scope_name, + collection_name=self.collection_name, + embedding=self.embedding, + index_name=self.index_name, + ) else: - documents.append(_input) - if documents: - vector_store = CouchbaseVectorStore.from_documents( - documents=documents, - cluster=cluster, - bucket_name=bucket_name, - scope_name=scope_name, - collection_name=collection_name, - embedding=embedding, - index_name=index_name, - ) + couchbase_vs = CouchbaseVectorStore( + cluster=cluster, + bucket_name=self.bucket_name, + scope_name=self.scope_name, + collection_name=self.collection_name, + embedding=self.embedding, + index_name=self.index_name, + ) else: - vector_store = CouchbaseVectorStore( + couchbase_vs = CouchbaseVectorStore( cluster=cluster, - bucket_name=bucket_name, - scope_name=scope_name, - collection_name=collection_name, - embedding=embedding, - index_name=index_name, + bucket_name=self.bucket_name, + scope_name=self.scope_name, + collection_name=self.collection_name, + embedding=self.embedding, + index_name=self.index_name, ) - return vector_store + + return couchbase_vs + + def search_documents(self) -> List[Data]: + vector_store = self._build_couchbase() + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + + data = docs_to_data(docs) + self.status = data + return data + else: + return [] diff --git a/src/backend/base/langflow/components/vectorstores/FAISS.py b/src/backend/base/langflow/components/vectorstores/FAISS.py index 0dd59a576..4f502f5b4 100644 --- a/src/backend/base/langflow/components/vectorstores/FAISS.py +++ b/src/backend/base/langflow/components/vectorstores/FAISS.py @@ -1,46 +1,184 @@ -from typing import List, Text, Union +from copy import deepcopy +from typing import List, Optional, Union -from langchain_community.vectorstores.faiss import FAISS -from langchain_core.retrievers import BaseRetriever -from langchain_core.vectorstores import VectorStore +from langchain.vectorstores import FAISS +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever -from langflow.custom import CustomComponent -from langflow.field_typing import Embeddings +from langflow.custom import Component +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput from langflow.schema import Data +from langflow.template import Output +from langflow.field_typing import Embeddings, Text +from langflow.helpers.data import docs_to_data + +from loguru import logger -class FAISSComponent(CustomComponent): - display_name = "FAISS" - description = "Ingest documents into FAISS Vector Store." +class FAISSVectorStoreComponent(Component): + """ + FAISS Vector Store with search capabilities + """ + + display_name: str = "FAISS" + description: str = "FAISS Vector Store with search capabilities" documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/faiss" + icon = "FAISS" - def build_config(self): - return { - "inputs": {"display_name": "Input", "input_types": ["Document", "Data"]}, - "embedding": {"display_name": "Embedding"}, - "folder_path": { - "display_name": "Folder Path", - "info": "Path to save the FAISS index. It will be relative to where Langflow is running.", - }, - "index_name": {"display_name": "Index Name"}, - } + inputs = [ + StrInput( + name="folder_path", + display_name="Folder Path", + info="Path to save the FAISS index. It will be relative to where Langflow is running.", + ), + StrInput( + name="index_name", + display_name="Index Name", + value="langflow_index", + ), + HandleInput( + name="embedding", display_name="Embedding", input_types=["Embeddings"] + ), + StrInput( + name="vector_store_inputs", + display_name="Vector Store Inputs", + input_types=["Document", "Data"], + is_list=True, + ), + BoolInput( + name="add_to_vector_store", + display_name="Add to Vector Store", + info="If true, the Vector Store Inputs will be added to the Vector Store.", + ), + BoolInput( + name="allow_dangerous_deserialization", + display_name="Allow Dangerous Deserialization", + info="Set to True to allow loading pickle files from untrusted sources. Only enable this if you trust the source of the data.", + advanced=True, + value=False, + ), + StrInput( + name="search_input", + display_name="Search Input", + ), + IntInput( + name="number_of_results", + display_name="Number of Results", + info="Number of results to return.", + advanced=True, + value=4, + ), + ] - def build( - self, - embedding: Embeddings, - inputs: List[Data], - folder_path: str, - index_name: str = "langflow_index", - ) -> Union[VectorStore, FAISS, BaseRetriever]: - documents = [] - for _input in inputs or []: - if isinstance(_input, Data): - documents.append(_input.to_lc_document()) - else: - documents.append(_input) - vector_store = FAISS.from_documents(documents=documents, embedding=embedding) - if not folder_path: + outputs = [ + Output( + display_name="Vector Store", + name="vector_store", + method="build_vector_store", + ), + Output( + display_name="Base Retriever", + name="base_retriever", + method="build_base_retriever", + ), + Output( + display_name="Search Results", + name="search_results", + method="search_documents", + ), + ] + + def build_vector_store(self) -> FAISS: + """ + Builds the Vector Store object. + """ + return self._build_faiss() + + def build_base_retriever(self) -> BaseRetriever: + """ + Builds the BaseRetriever object. + """ + return self._build_faiss() + + def _build_faiss(self) -> FAISS: + """ + Builds the FAISS object. + """ + if not self.folder_path: raise ValueError("Folder path is required to save the FAISS index.") - path = self.resolve_path(folder_path) - vector_store.save_local(Text(path), index_name) - return vector_store + path = self.resolve_path(self.folder_path) + + if self.add_to_vector_store: + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + documents.append(_input) + + faiss = FAISS.from_documents(documents=documents, embedding=self.embedding) + faiss.save_local(Text(path), self.index_name) + else: + try: + faiss = FAISS.load_local( + folder_path=Text(path), + embeddings=self.embedding, + index_name=self.index_name, + allow_dangerous_deserialization=self.allow_dangerous_deserialization, + ) + except Exception as e: + raise ValueError( + "Failed to load the FAISS index. Make sure the index was created with trusted data. " + "If you trust the data source, you can set `allow_dangerous_deserialization` to `True` " + "in the component's advanced settings to enable deserialization." + ) from e + + return faiss + + def search_documents(self) -> List[Data]: + """ + Search for documents in the FAISS vector store. + """ + if not self.folder_path: + raise ValueError("Folder path is required to load the FAISS index.") + path = self.resolve_path(self.folder_path) + + try: + vector_store = FAISS.load_local( + folder_path=Text(path), + embeddings=self.embedding, + index_name=self.index_name, + allow_dangerous_deserialization=self.allow_dangerous_deserialization, + ) + except Exception as e: + raise ValueError( + "Failed to load the FAISS index. Make sure the index was created with trusted data. " + "If you trust the data source, you can set `allow_dangerous_deserialization` to `True` " + "in the component's advanced settings to enable deserialization." + ) from e + + if not vector_store: + raise ValueError("Failed to load the FAISS index.") + + logger.debug(f"Search input: {self.search_input}") + logger.debug(f"Number of results: {self.number_of_results}") + + if ( + self.search_input + and isinstance(self.search_input, str) + and self.search_input.strip() + ): + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + + logger.debug(f"Retrieved documents: {len(docs)}") + + data = docs_to_data(docs) + logger.debug(f"Converted documents to data: {len(data)}") + logger.debug(data) + return data # Return the search results data + else: + logger.debug("No search input provided. Skipping search.") + return [] diff --git a/src/backend/base/langflow/components/vectorstores/MongoDBAtlasVector.py b/src/backend/base/langflow/components/vectorstores/MongoDBAtlasVector.py index c69931e25..32608b75d 100644 --- a/src/backend/base/langflow/components/vectorstores/MongoDBAtlasVector.py +++ b/src/backend/base/langflow/components/vectorstores/MongoDBAtlasVector.py @@ -1,64 +1,101 @@ from typing import List, Optional -from langchain_community.vectorstores.mongodb_atlas import MongoDBAtlasVectorSearch +from langchain_community.vectorstores import MongoDBAtlasVectorSearch +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever -from langflow.custom import CustomComponent -from langflow.field_typing import Embeddings +from langflow.custom import Component +from langflow.field_typing import Embeddings, Text from langflow.schema import Data +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput +from langflow.template import Output +from langflow.helpers.data import docs_to_data - -class MongoDBAtlasComponent(CustomComponent): +class MongoDBAtlasComponent(Component): display_name = "MongoDB Atlas" - description = "Construct a `MongoDB Atlas Vector Search` vector store from raw documents." + description = "MongoDB Atlas Vector Store with search capabilities" + documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/mongodb_atlas" icon = "MongoDB" - def build_config(self): - return { - "inputs": {"display_name": "Input", "input_types": ["Document", "Data"]}, - "embedding": {"display_name": "Embedding"}, - "collection_name": {"display_name": "Collection Name"}, - "db_name": {"display_name": "Database Name"}, - "index_name": {"display_name": "Index Name"}, - "mongodb_atlas_cluster_uri": {"display_name": "MongoDB Atlas Cluster URI"}, - } + inputs = [ + StrInput(name="mongodb_atlas_cluster_uri", display_name="MongoDB Atlas Cluster URI", required=True), + StrInput(name="db_name", display_name="Database Name", required=True), + StrInput(name="collection_name", display_name="Collection Name", required=True), + StrInput(name="index_name", display_name="Index Name", required=True), + HandleInput(name="embedding", display_name="Embedding", input_types=["Embeddings"]), + HandleInput(name="vector_store_inputs", display_name="Vector Store Inputs", input_types=["Document", "Data"], is_list=True), + BoolInput(name="add_to_vector_store", display_name="Add to Vector Store", info="If true, the Vector Store Inputs will be added to the Vector Store."), + StrInput(name="search_input", display_name="Search Input"), + IntInput(name="number_of_results", display_name="Number of Results", info="Number of results to return.", value=4, advanced=True), + ] - def build( - self, - embedding: Embeddings, - inputs: Optional[List[Data]] = None, - collection_name: str = "", - db_name: str = "", - index_name: str = "", - mongodb_atlas_cluster_uri: str = "", - ) -> MongoDBAtlasVectorSearch: + outputs = [ + Output(display_name="Vector Store", name="vector_store", method="build_vector_store", output_type=MongoDBAtlasVectorSearch), + Output(display_name="Base Retriever", name="base_retriever", method="build_base_retriever", output_type=BaseRetriever), + Output(display_name="Search Results", name="search_results", method="search_documents"), + ] + + def build_vector_store(self) -> MongoDBAtlasVectorSearch: + return self._build_mongodb_atlas() + + def build_base_retriever(self) -> BaseRetriever: + return self._build_mongodb_atlas() + + def _build_mongodb_atlas(self) -> MongoDBAtlasVectorSearch: try: from pymongo import MongoClient except ImportError: raise ImportError("Please install pymongo to use MongoDB Atlas Vector Store") + try: - mongo_client: MongoClient = MongoClient(mongodb_atlas_cluster_uri) - collection = mongo_client[db_name][collection_name] + mongo_client: MongoClient = MongoClient(self.mongodb_atlas_cluster_uri) + collection = mongo_client[self.db_name][self.collection_name] except Exception as e: raise ValueError(f"Failed to connect to MongoDB Atlas: {e}") - documents = [] - for _input in inputs or []: - if isinstance(_input, Data): - documents.append(_input.to_lc_document()) + + if self.add_to_vector_store: + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + documents.append(_input) + + if documents: + vector_store = MongoDBAtlasVectorSearch.from_documents( + documents=documents, + embedding=self.embedding, + collection=collection, + db_name=self.db_name, + index_name=self.index_name, + mongodb_atlas_cluster_uri=self.mongodb_atlas_cluster_uri, + ) else: - documents.append(_input) - if documents: - vector_store = MongoDBAtlasVectorSearch.from_documents( - documents=documents, - embedding=embedding, - collection=collection, - db_name=db_name, - index_name=index_name, - mongodb_atlas_cluster_uri=mongodb_atlas_cluster_uri, - ) + vector_store = MongoDBAtlasVectorSearch( + embedding=self.embedding, + collection=collection, + index_name=self.index_name, + ) else: vector_store = MongoDBAtlasVectorSearch( - embedding=embedding, + embedding=self.embedding, collection=collection, - index_name=index_name, + index_name=self.index_name, ) + return vector_store + + def search_documents(self) -> List[Data]: + vector_store = self._build_mongodb_atlas() + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + + data = docs_to_data(docs) + self.status = data + return data + else: + return [] diff --git a/src/backend/base/langflow/components/vectorstores/Pinecone.py b/src/backend/base/langflow/components/vectorstores/Pinecone.py index 1fa0b937f..138c9f77a 100644 --- a/src/backend/base/langflow/components/vectorstores/Pinecone.py +++ b/src/backend/base/langflow/components/vectorstores/Pinecone.py @@ -1,151 +1,87 @@ from typing import List, Optional, Union -from langchain_core.documents import Document -from langchain_core.retrievers import BaseRetriever -from langchain_core.vectorstores import VectorStore -from langchain_pinecone._utilities import DistanceStrategy -from langchain_pinecone.vectorstores import PineconeVectorStore +from langchain.vectorstores import Pinecone +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever -from langflow.custom import CustomComponent -from langflow.field_typing import Embeddings +from langflow.custom import Component +from langflow.field_typing import Embeddings, Text from langflow.schema import Data +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput, DropdownInput +from langflow.template import Output +from langflow.helpers.data import docs_to_data - -class PineconeComponent(CustomComponent): +class PineconeVectorStoreComponent(Component): display_name = "Pinecone" - description = "Construct Pinecone wrapper from raw documents." + description = "Pinecone Vector Store with search capabilities" + documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/pinecone" icon = "Pinecone" - field_order = ["index_name", "namespace", "distance_strategy", "pinecone_api_key", "documents", "embedding"] - def build_config(self): - distance_options = [e.value.title().replace("_", " ") for e in DistanceStrategy] - distance_value = distance_options[0] - return { - "inputs": {"display_name": "Input", "input_types": ["Document", "Data"]}, - "embedding": {"display_name": "Embedding"}, - "index_name": {"display_name": "Index Name"}, - "namespace": {"display_name": "Namespace"}, - "text_key": {"display_name": "Text Key"}, - "distance_strategy": { - "display_name": "Distance Strategy", - # get values from enum - # and make them title case for display - "options": distance_options, - "advanced": True, - "value": distance_value, - }, - "pinecone_api_key": { - "display_name": "Pinecone API Key", - "default": "", - "password": True, - "required": True, - }, - "pool_threads": { - "display_name": "Pool Threads", - "default": 1, - "advanced": True, - }, - } + inputs = [ + StrInput(name="index_name", display_name="Index Name", required=True), + StrInput(name="namespace", display_name="Namespace", info="Namespace for the index."), + DropdownInput(name="distance_strategy", display_name="Distance Strategy", options=["Cosine", "Euclidean", "Dot Product"], value="Cosine", advanced=True), + StrInput(name="pinecone_api_key", display_name="Pinecone API Key", password=True, required=True), + HandleInput(name="embedding", display_name="Embedding", input_types=["Embeddings"]), + StrInput(name="text_key", display_name="Text Key", info="Key in the record to use as text.", value="text", advanced=True), + HandleInput(name="vector_store_inputs", display_name="Vector Store Inputs", input_types=["Document", "Data"], is_list=True), + BoolInput(name="add_to_vector_store", display_name="Add to Vector Store", info="If true, the Vector Store Inputs will be added to the Vector Store."), + StrInput(name="search_input", display_name="Search Input"), + IntInput(name="number_of_results", display_name="Number of Results", info="Number of results to return.", value=4, advanced=True), + ] - def from_existing_index( - self, - index_name: str, - embedding: Embeddings, - pinecone_api_key: str | None, - text_key: str = "text", - namespace: Optional[str] = None, - distance_strategy: DistanceStrategy = DistanceStrategy.COSINE, - pool_threads: int = 4, - ) -> PineconeVectorStore: - """Load pinecone vectorstore from index name.""" - pinecone_index = PineconeVectorStore.get_pinecone_index( - index_name, pool_threads, pinecone_api_key=pinecone_api_key - ) - return PineconeVectorStore( - index=pinecone_index, - embedding=embedding, - text_key=text_key, - namespace=namespace, - distance_strategy=distance_strategy, + outputs = [ + Output(display_name="Vector Store", name="vector_store", method="build_vector_store", output_type=Pinecone), + Output(display_name="Base Retriever", name="base_retriever", method="build_base_retriever", output_type=BaseRetriever), + Output(display_name="Search Results", name="search_results", method="search_documents"), + ] + + def build_vector_store(self) -> Pinecone: + return self._build_pinecone() + + def build_base_retriever(self) -> BaseRetriever: + return self._build_pinecone() + + def _build_pinecone(self) -> Pinecone: + from langchain_pinecone._utilities import DistanceStrategy + from langchain_pinecone.vectorstores import Pinecone + + distance_strategy = self.distance_strategy.replace(" ", "_").upper() + _distance_strategy = DistanceStrategy[distance_strategy] + + pinecone = Pinecone( + index_name=self.index_name, + embedding=self.embedding, + text_key=self.text_key, + namespace=self.namespace, + distance_strategy=_distance_strategy, + pinecone_api_key=self.pinecone_api_key, ) - def from_documents( - self, - documents: List[Document], - embedding: Embeddings, - index_name: str, - pinecone_api_key: str | None, - text_key: str = "text", - namespace: Optional[str] = None, - pool_threads: int = 4, - distance_strategy: DistanceStrategy = DistanceStrategy.COSINE, - batch_size: int = 32, - upsert_kwargs: Optional[dict] = None, - embeddings_chunk_size: int = 1000, - ) -> PineconeVectorStore: - """Create a new pinecone vectorstore from documents.""" - texts = [d.page_content for d in documents] - metadatas = [d.metadata for d in documents] - pinecone = self.from_existing_index( - index_name=index_name, - embedding=embedding, - pinecone_api_key=pinecone_api_key, - text_key=text_key, - namespace=namespace, - distance_strategy=distance_strategy, - pool_threads=pool_threads, - ) - pinecone.add_texts( - texts, - metadatas=metadatas, - ids=None, - namespace=namespace, - batch_size=batch_size, - embedding_chunk_size=embeddings_chunk_size, - **(upsert_kwargs or {}), - ) + if self.add_to_vector_store: + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + documents.append(_input) + + if documents: + pinecone.add_documents(documents) + return pinecone - def build( - self, - embedding: Embeddings, - distance_strategy: str, - inputs: Optional[List[Data]] = None, - text_key: str = "text", - pool_threads: int = 4, - index_name: Optional[str] = None, - pinecone_api_key: Optional[str] = None, - namespace: Optional[str] = "default", - ) -> Union[VectorStore, BaseRetriever]: - # get distance strategy from string - distance_strategy = distance_strategy.replace(" ", "_").upper() - _distance_strategy = DistanceStrategy[distance_strategy] - if not index_name: - raise ValueError("Index Name is required.") - documents = [] - for _input in inputs or []: - if isinstance(_input, Data): - documents.append(_input.to_lc_document()) - else: - documents.append(_input) - if documents: - return self.from_documents( - documents=documents, - embedding=embedding, - index_name=index_name, - pinecone_api_key=pinecone_api_key, - text_key=text_key, - namespace=namespace, - distance_strategy=_distance_strategy, - pool_threads=pool_threads, + def search_documents(self) -> List[Data]: + vector_store = self._build_pinecone() + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, ) - return self.from_existing_index( - index_name=index_name, - embedding=embedding, - pinecone_api_key=pinecone_api_key, - text_key=text_key, - namespace=namespace, - distance_strategy=_distance_strategy, - pool_threads=pool_threads, - ) + data = docs_to_data(docs) + self.status = data + return data + else: + return [] diff --git a/src/backend/base/langflow/components/vectorstores/Qdrant.py b/src/backend/base/langflow/components/vectorstores/Qdrant.py index 7b63cebb6..98a7a6938 100644 --- a/src/backend/base/langflow/components/vectorstores/Qdrant.py +++ b/src/backend/base/langflow/components/vectorstores/Qdrant.py @@ -1,114 +1,115 @@ -from typing import Optional, Union +from typing import List, Optional -from langchain_community.vectorstores.qdrant import Qdrant -from langchain_core.retrievers import BaseRetriever -from langchain_core.vectorstores import VectorStore +from langchain.vectorstores import Qdrant +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever -from langflow.custom import CustomComponent -from langflow.field_typing import Embeddings +from langflow.custom import Component +from langflow.field_typing import Embeddings, Text from langflow.schema import Data +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput, DropdownInput +from langflow.template import Output +from langflow.helpers.data import docs_to_data - -class QdrantComponent(CustomComponent): +class QdrantVectorStoreComponent(Component): display_name = "Qdrant" - description = "Construct Qdrant wrapper from a list of texts." + description = "Qdrant Vector Store with search capabilities" + documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/qdrant" icon = "Qdrant" - def build_config(self): - return { - "inputs": {"display_name": "Input", "input_types": ["Document", "Data"]}, - "embedding": {"display_name": "Embedding"}, - "api_key": {"display_name": "API Key", "password": True, "advanced": True}, - "collection_name": {"display_name": "Collection Name"}, - "content_payload_key": { - "display_name": "Content Payload Key", - "advanced": True, - }, - "distance_func": {"display_name": "Distance Function", "advanced": True}, - "grpc_port": {"display_name": "gRPC Port", "advanced": True}, - "host": {"display_name": "Host", "advanced": True}, - "https": {"display_name": "HTTPS", "advanced": True}, - "location": {"display_name": "Location", "advanced": True}, - "metadata_payload_key": { - "display_name": "Metadata Payload Key", - "advanced": True, - }, - "path": {"display_name": "Path", "advanced": True}, - "port": {"display_name": "Port", "advanced": True}, - "prefer_grpc": {"display_name": "Prefer gRPC", "advanced": True}, - "prefix": {"display_name": "Prefix", "advanced": True}, - "timeout": {"display_name": "Timeout", "advanced": True}, - "url": {"display_name": "URL", "advanced": True}, + inputs = [ + StrInput(name="collection_name", display_name="Collection Name", required=True), + StrInput(name="host", display_name="Host", value="localhost", advanced=True), + IntInput(name="port", display_name="Port", value=6333, advanced=True), + IntInput(name="grpc_port", display_name="gRPC Port", value=6334, advanced=True), + StrInput(name="api_key", display_name="API Key", password=True, advanced=True), + StrInput(name="prefix", display_name="Prefix", advanced=True), + IntInput(name="timeout", display_name="Timeout", advanced=True), + StrInput(name="path", display_name="Path", advanced=True), + StrInput(name="url", display_name="URL", advanced=True), + DropdownInput(name="distance_func", display_name="Distance Function", options=["Cosine", "Euclidean", "Dot Product"], value="Cosine", advanced=True), + StrInput(name="content_payload_key", display_name="Content Payload Key", value="page_content", advanced=True), + StrInput(name="metadata_payload_key", display_name="Metadata Payload Key", value="metadata", advanced=True), + HandleInput(name="embedding", display_name="Embedding", input_types=["Embeddings"]), + HandleInput(name="vector_store_inputs", display_name="Vector Store Inputs", input_types=["Document", "Data"], is_list=True), + BoolInput(name="add_to_vector_store", display_name="Add to Vector Store", info="If true, the Vector Store Inputs will be added to the Vector Store."), + StrInput(name="search_input", display_name="Search Input"), + IntInput(name="number_of_results", display_name="Number of Results", info="Number of results to return.", value=4, advanced=True), + ] + + outputs = [ + Output(display_name="Vector Store", name="vector_store", method="build_vector_store", output_type=Qdrant), + Output(display_name="Base Retriever", name="base_retriever", method="build_base_retriever", output_type=BaseRetriever), + Output(display_name="Search Results", name="search_results", method="search_documents"), + ] + + def build_vector_store(self) -> Qdrant: + return self._build_qdrant() + + def build_base_retriever(self) -> BaseRetriever: + return self._build_qdrant() + + def _build_qdrant(self) -> Qdrant: + qdrant_kwargs = { + "collection_name": self.collection_name, + "content_payload_key": self.content_payload_key, + "distance_func": self.distance_func, + "metadata_payload_key": self.metadata_payload_key, } - def build( - self, - embedding: Embeddings, - collection_name: str, - inputs: Optional[Data] = None, - api_key: Optional[str] = None, - content_payload_key: str = "page_content", - distance_func: str = "Cosine", - grpc_port: int = 6334, - https: bool = False, - host: Optional[str] = None, - location: Optional[str] = None, - metadata_payload_key: str = "metadata", - path: Optional[str] = None, - port: Optional[int] = 6333, - prefer_grpc: bool = False, - prefix: Optional[str] = None, - timeout: Optional[int] = None, - url: Optional[str] = None, - ) -> Union[VectorStore, Qdrant, BaseRetriever]: - documents = [] - for _input in inputs or []: - if isinstance(_input, Data): - documents.append(_input.to_lc_document()) + server_kwargs = { + "host": self.host, + "port": self.port, + "grpc_port": self.grpc_port, + "api_key": self.api_key, + "prefix": self.prefix, + "timeout": self.timeout, + "path": self.path, + "url": self.url, + } + + # Remove None values from server_kwargs + server_kwargs = {k: v for k, v in server_kwargs.items() if v is not None} + + if self.add_to_vector_store: + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + documents.append(_input) + + if documents: + qdrant = Qdrant.from_documents( + documents, + embedding=self.embedding, + client_kwargs=server_kwargs, + **qdrant_kwargs + ) else: - documents.append(_input) - if not documents: + from qdrant_client import QdrantClient + + client = QdrantClient(**server_kwargs) + qdrant = Qdrant(embedding_function=self.embedding.embed_query, client=client, **qdrant_kwargs) + else: from qdrant_client import QdrantClient - client = QdrantClient( - location=location, - url=url, - port=port, - grpc_port=grpc_port, - https=https, - prefix=prefix, - timeout=timeout, - prefer_grpc=prefer_grpc, - api_key=api_key, - host=host, - path=path, + client = QdrantClient(**server_kwargs) + qdrant = Qdrant(embedding_function=self.embedding.embed_query, client=client, **qdrant_kwargs) + + return qdrant + + def search_documents(self) -> List[Data]: + vector_store = self._build_qdrant() + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, ) - vs = Qdrant( - client=client, - collection_name=collection_name, - embeddings=embedding, - content_payload_key=content_payload_key, - metadata_payload_key=metadata_payload_key, - ) - return vs + + data = docs_to_data(docs) + self.status = data + return data else: - vs = Qdrant.from_documents( - documents=documents, # type: ignore - embedding=embedding, - api_key=api_key, - collection_name=collection_name, - content_payload_key=content_payload_key, - distance_func=distance_func, - grpc_port=grpc_port, - host=host, - https=https, - location=location, - metadata_payload_key=metadata_payload_key, - path=path, - port=port, - prefer_grpc=prefer_grpc, - prefix=prefix, - timeout=timeout, - url=url, - ) - return vs + return [] diff --git a/src/backend/base/langflow/components/vectorstores/SupabaseVectorStore.py b/src/backend/base/langflow/components/vectorstores/SupabaseVectorStore.py index a9ca62452..d928ccb09 100644 --- a/src/backend/base/langflow/components/vectorstores/SupabaseVectorStore.py +++ b/src/backend/base/langflow/components/vectorstores/SupabaseVectorStore.py @@ -1,49 +1,85 @@ from typing import List, Optional, Union -from langchain_community.vectorstores.supabase import SupabaseVectorStore -from langchain_core.retrievers import BaseRetriever -from langchain_core.vectorstores import VectorStore +from langchain_community.vectorstores import SupabaseVectorStore +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever from supabase.client import Client, create_client -from langflow.custom import CustomComponent -from langflow.field_typing import Embeddings +from langflow.custom import Component +from langflow.field_typing import Embeddings, Text from langflow.schema import Data +from langflow.inputs import IntInput, StrInput, HandleInput +from langflow.template import Output +from langflow.helpers.data import docs_to_data - -class SupabaseComponent(CustomComponent): +class SupabaseVectorStoreComponent(Component): display_name = "Supabase" - description = "Return VectorStore initialized from texts and embeddings." + description = "Supabase Vector Store with search capabilities" + documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/supabase" + icon = "Supabase" - def build_config(self): - return { - "inputs": {"display_name": "Input", "input_types": ["Document", "Data"]}, - "embedding": {"display_name": "Embedding"}, - "query_name": {"display_name": "Query Name"}, - "supabase_service_key": {"display_name": "Supabase Service Key"}, - "supabase_url": {"display_name": "Supabase URL"}, - "table_name": {"display_name": "Table Name", "advanced": True}, - } + inputs = [ + StrInput(name="supabase_url", display_name="Supabase URL", required=True), + StrInput(name="supabase_service_key", display_name="Supabase Service Key", required=True), + StrInput(name="table_name", display_name="Table Name", advanced=True), + StrInput(name="query_name", display_name="Query Name"), + HandleInput(name="embedding", display_name="Embedding", input_types=["Embeddings"]), + HandleInput(name="vector_store_inputs", display_name="Vector Store Inputs", input_types=["Document", "Data"], is_list=True), + StrInput(name="search_input", display_name="Search Input"), + IntInput(name="number_of_results", display_name="Number of Results", info="Number of results to return.", value=4, advanced=True), + ] + + outputs = [ + Output(display_name="Vector Store", name="vector_store", method="build_vector_store", output_type=SupabaseVectorStore), + Output(display_name="Base Retriever", name="base_retriever", method="build_base_retriever", output_type=BaseRetriever), + Output(display_name="Search Results", name="search_results", method="search_documents"), + ] + + def build_vector_store(self) -> SupabaseVectorStore: + return self._build_supabase() + + def build_base_retriever(self) -> BaseRetriever: + return self._build_supabase() + + def _build_supabase(self) -> SupabaseVectorStore: + supabase: Client = create_client(self.supabase_url, supabase_key=self.supabase_service_key) - def build( - self, - embedding: Embeddings, - inputs: Optional[List[Data]] = None, - query_name: str = "", - supabase_service_key: str = "", - supabase_url: str = "", - table_name: str = "", - ) -> Union[VectorStore, SupabaseVectorStore, BaseRetriever]: - supabase: Client = create_client(supabase_url, supabase_key=supabase_service_key) documents = [] - for _input in inputs or []: + for _input in self.vector_store_inputs or []: if isinstance(_input, Data): documents.append(_input.to_lc_document()) else: documents.append(_input) - return SupabaseVectorStore.from_documents( - documents=documents, - embedding=embedding, - query_name=query_name, - client=supabase, - table_name=table_name, - ) + + if documents: + supabase_vs = SupabaseVectorStore.from_documents( + documents=documents, + embedding=self.embedding, + query_name=self.query_name, + client=supabase, + table_name=self.table_name, + ) + else: + supabase_vs = SupabaseVectorStore( + client=supabase, + embedding=self.embedding, + table_name=self.table_name, + query_name=self.query_name, + ) + + return supabase_vs + + def search_documents(self) -> List[Data]: + vector_store = self._build_supabase() + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + + data = docs_to_data(docs) + self.status = data + return data + else: + return [] diff --git a/src/backend/base/langflow/components/vectorstores/Upstash.py b/src/backend/base/langflow/components/vectorstores/Upstash.py index 6720d9b9e..dfd485eea 100644 --- a/src/backend/base/langflow/components/vectorstores/Upstash.py +++ b/src/backend/base/langflow/components/vectorstores/Upstash.py @@ -1,89 +1,101 @@ from typing import List, Optional, Union -from langchain_community.vectorstores.upstash import UpstashVectorStore -from langchain_core.embeddings import Embeddings -from langchain_core.retrievers import BaseRetriever -from langchain_core.vectorstores import VectorStore +from langchain_community.vectorstores import UpstashVectorStore +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever -from langflow.custom import CustomComponent +from langflow.custom import Component +from langflow.field_typing import Embeddings, Text from langflow.schema import Data +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput +from langflow.template import Output +from langflow.helpers.data import docs_to_data +class UpstashVectorStoreComponent(Component): + display_name = "Upstash" + description = "Upstash Vector Store with search capabilities" + documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/upstash" + icon = "Upstash" -class UpstashVectorStoreComponent(CustomComponent): - """ - A custom component for implementing a Vector Store using Upstash. - """ + inputs = [ + StrInput(name="index_url", display_name="Index URL", info="The URL of the Upstash index.", required=True), + StrInput(name="index_token", display_name="Index Token", info="The token for the Upstash index.", required=True), + StrInput(name="text_key", display_name="Text Key", info="The key in the record to use as text.", value="text", advanced=True), + HandleInput(name="embedding", display_name="Embedding", input_types=["Embeddings"], info="To use Upstash's embeddings, don't provide an embedding."), + HandleInput(name="vector_store_inputs", display_name="Vector Store Inputs", input_types=["Document", "Data"], is_list=True), + BoolInput(name="add_to_vector_store", display_name="Add to Vector Store", info="If true, the Vector Store Inputs will be added to the Vector Store."), + StrInput(name="search_input", display_name="Search Input"), + IntInput(name="number_of_results", display_name="Number of Results", info="Number of results to return.", value=4, advanced=True), + ] - display_name: str = "Upstash" - description: str = "Create and Utilize an Upstash Vector Store" + outputs = [ + Output(display_name="Vector Store", name="vector_store", method="build_vector_store", output_type=UpstashVectorStore), + Output(display_name="Base Retriever", name="base_retriever", method="build_base_retriever", output_type=BaseRetriever), + Output(display_name="Search Results", name="search_results", method="search_documents"), + ] - def build_config(self): - """ - Builds the configuration for the component. + def build_vector_store(self) -> UpstashVectorStore: + return self._build_upstash() - Returns: - - dict: A dictionary containing the configuration options for the component. - """ - return { - "inputs": {"display_name": "Input", "input_types": ["Document", "Data"]}, - "embedding": { - "display_name": "Embedding", - "input_types": ["Embeddings"], - "info": "To use Upstash's embeddings, don't provide an embedding.", - }, - "index_url": { - "display_name": "Index URL", - "info": "The URL of the Upstash index.", - }, - "index_token": { - "display_name": "Index Token", - "info": "The token for the Upstash index.", - }, - "text_key": { - "display_name": "Text Key", - "info": "The key in the record to use as text.", - "advanced": True, - }, - } + def build_base_retriever(self) -> BaseRetriever: + return self._build_upstash() - def build( - self, - inputs: Optional[List[Data]] = None, - text_key: str = "text", - index_url: Optional[str] = None, - index_token: Optional[str] = None, - embedding: Optional[Embeddings] = None, - ) -> Union[VectorStore, BaseRetriever]: - documents = [] - for _input in inputs or []: - if isinstance(_input, Data): - documents.append(_input.to_lc_document()) + def _build_upstash(self) -> UpstashVectorStore: + use_upstash_embedding = self.embedding is None + + if self.add_to_vector_store: + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + documents.append(_input) + + if documents: + if use_upstash_embedding: + upstash_vs = UpstashVectorStore( + embedding=use_upstash_embedding, + text_key=self.text_key, + index_url=self.index_url, + index_token=self.index_token, + ) + upstash_vs.add_documents(documents) + else: + upstash_vs = UpstashVectorStore.from_documents( + documents=documents, + embedding=self.embedding, + text_key=self.text_key, + index_url=self.index_url, + index_token=self.index_token, + ) else: - documents.append(_input) - - use_upstash_embedding = embedding is None - if not documents: - upstash_vs = UpstashVectorStore( - embedding=embedding or use_upstash_embedding, - text_key=text_key, - index_url=index_url, - index_token=index_token, - ) - else: - if use_upstash_embedding: upstash_vs = UpstashVectorStore( - embedding=use_upstash_embedding, - text_key=text_key, - index_url=index_url, - index_token=index_token, - ) - upstash_vs.add_documents(documents) - elif embedding: - upstash_vs = UpstashVectorStore.from_documents( - documents=documents, # type: ignore - embedding=embedding, - text_key=text_key, - index_url=index_url, - index_token=index_token, + embedding=self.embedding or use_upstash_embedding, + text_key=self.text_key, + index_url=self.index_url, + index_token=self.index_token, ) + else: + upstash_vs = UpstashVectorStore( + embedding=self.embedding or use_upstash_embedding, + text_key=self.text_key, + index_url=self.index_url, + index_token=self.index_token, + ) + return upstash_vs + + def search_documents(self) -> List[Data]: + vector_store = self._build_upstash() + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + + data = docs_to_data(docs) + self.status = data + return data + else: + return [] diff --git a/src/backend/base/langflow/components/vectorstores/Vectara.py b/src/backend/base/langflow/components/vectorstores/Vectara.py index f5d5253fa..a2635a2e3 100644 --- a/src/backend/base/langflow/components/vectorstores/Vectara.py +++ b/src/backend/base/langflow/components/vectorstores/Vectara.py @@ -1,90 +1,86 @@ import tempfile import urllib import urllib.request -from typing import List, Optional, Union +from typing import List, Optional -from langchain_community.embeddings import FakeEmbeddings -from langchain_community.vectorstores.vectara import Vectara -from langchain_core.vectorstores import VectorStore +from langchain.embeddings import FakeEmbeddings +from langchain.vectorstores import Vectara +from langchain.schema import BaseRetriever -from langflow.custom import CustomComponent -from langflow.field_typing import BaseRetriever +from langflow.custom import Component +from langflow.field_typing import Embeddings, Text from langflow.schema import Data +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput +from langflow.template import Output +from langflow.helpers.data import docs_to_data - -class VectaraComponent(CustomComponent): - display_name: str = "Vectara" - description: str = "Implementation of Vector Store using Vectara" - documentation = "https://python.langchain.com/docs/integrations/vectorstores/vectara" +class VectaraVectorStoreComponent(Component): + display_name = "Vectara" + description = "Vectara Vector Store with search capabilities" + documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/vectara" icon = "Vectara" - field_config = { - "vectara_customer_id": { - "display_name": "Vectara Customer ID", - }, - "vectara_corpus_id": { - "display_name": "Vectara Corpus ID", - }, - "vectara_api_key": { - "display_name": "Vectara API Key", - "password": True, - }, - "inputs": { - "display_name": "Input", - "input_types": ["Document", "Data"], - "info": "If provided, will be upserted to corpus (optional)", - }, - "files_url": { - "display_name": "Files Url", - "info": "Make vectara object using url of files (optional)", - }, - } - def build( - self, - vectara_customer_id: str, - vectara_corpus_id: str, - vectara_api_key: str, - files_url: Optional[List[str]] = None, - inputs: Optional[Data] = None, - ) -> Union[VectorStore, BaseRetriever]: + inputs = [ + StrInput(name="vectara_customer_id", display_name="Vectara Customer ID", required=True), + StrInput(name="vectara_corpus_id", display_name="Vectara Corpus ID", required=True), + StrInput(name="vectara_api_key", display_name="Vectara API Key", password=True, required=True), + HandleInput(name="vector_store_inputs", display_name="Vector Store Inputs", input_types=["Document", "Data"], is_list=True), + BoolInput(name="add_to_vector_store", display_name="Add to Vector Store", info="If true, the Vector Store Inputs will be added to the Vector Store."), + StrInput(name="search_input", display_name="Search Input"), + IntInput(name="number_of_results", display_name="Number of Results", info="Number of results to return.", value=4, advanced=True), + ] + + outputs = [ + Output(display_name="Vector Store", name="vector_store", method="build_vector_store", output_type=Vectara), + Output(display_name="Base Retriever", name="base_retriever", method="build_base_retriever", output_type=BaseRetriever), + Output(display_name="Search Results", name="search_results", method="search_documents"), + ] + + def build_vector_store(self) -> Vectara: + return self._build_vectara() + + def build_base_retriever(self) -> BaseRetriever: + return self._build_vectara() + + def _build_vectara(self) -> Vectara: source = "Langflow" - documents = [] - for _input in inputs or []: - if isinstance(_input, Data): - documents.append(_input.to_lc_document()) - else: - documents.append(_input) + if self.add_to_vector_store: + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + documents.append(_input) - if documents: - return Vectara.from_documents( - documents=documents, # type: ignore - embedding=FakeEmbeddings(size=768), - vectara_customer_id=vectara_customer_id, - vectara_corpus_id=vectara_corpus_id, - vectara_api_key=vectara_api_key, - source=source, - ) - - if files_url is not None: - files_list = [] - for url in files_url: - name = tempfile.NamedTemporaryFile().name - urllib.request.urlretrieve(url, name) - files_list.append(name) - - return Vectara.from_files( - files=files_list, - embedding=FakeEmbeddings(size=768), - vectara_customer_id=vectara_customer_id, - vectara_corpus_id=vectara_corpus_id, - vectara_api_key=vectara_api_key, - source=source, - ) + if documents: + return Vectara.from_documents( + documents=documents, + embedding=FakeEmbeddings(size=768), + customer_id=self.vectara_customer_id, + corpus_id=self.vectara_corpus_id, + api_key=self.vectara_api_key, + source=source, + ) return Vectara( - vectara_customer_id=vectara_customer_id, - vectara_corpus_id=vectara_corpus_id, - vectara_api_key=vectara_api_key, + customer_id=self.vectara_customer_id, + corpus_id=self.vectara_corpus_id, + api_key=self.vectara_api_key, source=source, ) + + def search_documents(self) -> List[Data]: + vector_store = self._build_vectara() + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + + data = docs_to_data(docs) + self.status = data + return data + else: + return [] diff --git a/src/backend/base/langflow/components/vectorstores/Weaviate.py b/src/backend/base/langflow/components/vectorstores/Weaviate.py index c77ccb2a8..8cb5c5278 100644 --- a/src/backend/base/langflow/components/vectorstores/Weaviate.py +++ b/src/backend/base/langflow/components/vectorstores/Weaviate.py @@ -1,108 +1,91 @@ -from typing import Optional, Union +from typing import List, Optional, Union -import weaviate # type: ignore -from langchain_community.vectorstores import Weaviate -from langchain_core.documents import Document -from langchain_core.embeddings import Embeddings -from langchain_core.retrievers import BaseRetriever -from langchain_core.vectorstores import VectorStore +import weaviate +from langchain.vectorstores import Weaviate +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever -from langflow.custom import CustomComponent +from langflow.custom import Component +from langflow.field_typing import Embeddings, Text from langflow.schema import Data +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput +from langflow.template import Output +from langflow.helpers.data import docs_to_data +class WeaviateVectorStoreComponent(Component): + display_name = "Weaviate" + description = "Weaviate Vector Store with search capabilities" + documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/weaviate" + icon = "Weaviate" -class WeaviateVectorStoreComponent(CustomComponent): - display_name: str = "Weaviate" - description: str = "Implementation of Vector Store using Weaviate" - documentation = "https://python.langchain.com/docs/integrations/vectorstores/weaviate" - field_config = { - "url": {"display_name": "Weaviate URL", "value": "http://localhost:8080"}, - "api_key": { - "display_name": "API Key", - "password": True, - "required": False, - }, - "index_name": { - "display_name": "Index name", - "required": False, - }, - "text_key": { - "display_name": "Text Key", - "required": False, - "advanced": True, - "value": "text", - }, - "inputs": {"display_name": "Input", "input_types": ["Document", "Data"]}, - "embedding": {"display_name": "Embedding"}, - "attributes": { - "display_name": "Attributes", - "required": False, - "is_list": True, - "field_type": "str", - "advanced": True, - }, - "search_by_text": { - "display_name": "Search By Text", - "field_type": "bool", - "advanced": True, - }, - "code": {"show": False}, - } + inputs = [ + StrInput(name="url", display_name="Weaviate URL", value="http://localhost:8080", required=True), + StrInput(name="api_key", display_name="API Key", password=True, required=False), + StrInput(name="index_name", display_name="Index Name", required=True), + StrInput(name="text_key", display_name="Text Key", value="text", advanced=True), + HandleInput(name="embedding", display_name="Embedding", input_types=["Embeddings"]), + HandleInput(name="vector_store_inputs", display_name="Vector Store Inputs", input_types=["Document", "Data"], is_list=True), + BoolInput(name="add_to_vector_store", display_name="Add to Vector Store", info="If true, the Vector Store Inputs will be added to the Vector Store."), + StrInput(name="search_input", display_name="Search Input"), + IntInput(name="number_of_results", display_name="Number of Results", info="Number of results to return.", value=4, advanced=True), + BoolInput(name="search_by_text", display_name="Search By Text", advanced=True), + ] - def build( - self, - url: str, - index_name: str, - search_by_text: bool = False, - api_key: Optional[str] = None, - text_key: str = "text", - embedding: Optional[Embeddings] = None, - inputs: Optional[Data] = None, - attributes: Optional[list] = None, - ) -> Union[VectorStore, BaseRetriever]: - if api_key: - auth_config = weaviate.AuthApiKey(api_key=api_key) - client = weaviate.Client(url=url, auth_client_secret=auth_config) + outputs = [ + Output(display_name="Vector Store", name="vector_store", method="build_vector_store", output_type=Weaviate), + Output(display_name="Base Retriever", name="base_retriever", method="build_base_retriever", output_type=BaseRetriever), + Output(display_name="Search Results", name="search_results", method="search_documents"), + ] + + def build_vector_store(self) -> Weaviate: + return self._build_weaviate() + + def build_base_retriever(self) -> BaseRetriever: + return self._build_weaviate() + + def _build_weaviate(self) -> Weaviate: + if self.api_key: + auth_config = weaviate.AuthApiKey(api_key=self.api_key) + client = weaviate.Client(url=self.url, auth_client_secret=auth_config) else: - client = weaviate.Client(url=url) + client = weaviate.Client(url=self.url) - def _to_pascal_case(word: str): - if word and not word[0].isupper(): - word = word.capitalize() + if self.add_to_vector_store: + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + documents.append(_input) - if word.isidentifier(): - return word - - word = word.replace("-", " ").replace("_", " ") - parts = word.split() - pascal_case_word = "".join([part.capitalize() for part in parts]) - - return pascal_case_word - - index_name = _to_pascal_case(index_name) if index_name else None - if not index_name: - raise ValueError("Index name is required") - documents: list[Document] = [] - for _input in inputs or []: - if isinstance(_input, Data): - documents.append(_input.to_lc_document()) - elif isinstance(_input, Document): - documents.append(_input) - - if documents and embedding is not None: - return Weaviate.from_documents( - client=client, - index_name=index_name, - documents=documents, - embedding=embedding, - by_text=search_by_text, - ) + if documents and self.embedding: + return Weaviate.from_documents( + client=client, + index_name=self.index_name, + documents=documents, + embedding=self.embedding, + by_text=self.search_by_text, + ) return Weaviate( client=client, - index_name=index_name, - text_key=text_key, - embedding=embedding, - by_text=search_by_text, - attributes=attributes if attributes is not None else [], + index_name=self.index_name, + text_key=self.text_key, + embedding=self.embedding, + by_text=self.search_by_text, ) + + def search_documents(self) -> List[Data]: + vector_store = self._build_weaviate() + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + + data = docs_to_data(docs) + self.status = data + return data + else: + return [] diff --git a/src/backend/base/langflow/components/vectorstores/pgvector.py b/src/backend/base/langflow/components/vectorstores/pgvector.py index 36bb6f505..8658cb8bd 100644 --- a/src/backend/base/langflow/components/vectorstores/pgvector.py +++ b/src/backend/base/langflow/components/vectorstores/pgvector.py @@ -1,81 +1,86 @@ -from typing import Optional, Union +from typing import List, Optional, Union -from langchain_community.vectorstores.pgvector import PGVector -from langchain_core.embeddings import Embeddings -from langchain_core.retrievers import BaseRetriever -from langchain_core.vectorstores import VectorStore +from langchain_community.vectorstores import PGVector +from langchain.embeddings.base import Embeddings +from langchain.schema import BaseRetriever -from langflow.custom import CustomComponent +from langflow.custom import Component +from langflow.field_typing import Embeddings, Text from langflow.schema import Data +from langflow.inputs import BoolInput, IntInput, StrInput, HandleInput +from langflow.template import Output +from langflow.helpers.data import docs_to_data +class PGVectorComponent(Component): + display_name = "PGVector" + description = "PGVector Vector Store with search capabilities" + documentation = "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/pgvector" + icon = "PGVector" -class PGVectorComponent(CustomComponent): - """ - A custom component for implementing a Vector Store using PostgreSQL. - """ + inputs = [ + StrInput(name="pg_server_url", display_name="PostgreSQL Server Connection String", required=True), + StrInput(name="collection_name", display_name="Table", required=True), + HandleInput(name="embedding", display_name="Embedding", input_types=["Embeddings"]), + HandleInput(name="vector_store_inputs", display_name="Vector Store Inputs", input_types=["Document", "Data"], is_list=True), + BoolInput(name="add_to_vector_store", display_name="Add to Vector Store", info="If true, the Vector Store Inputs will be added to the Vector Store."), + StrInput(name="search_input", display_name="Search Input"), + IntInput(name="number_of_results", display_name="Number of Results", info="Number of results to return.", value=4, advanced=True), + ] - display_name: str = "PGVector" - description: str = "Implementation of Vector Store using PostgreSQL" - documentation = "https://python.langchain.com/docs/integrations/vectorstores/pgvector" + outputs = [ + Output(display_name="Vector Store", name="vector_store", method="build_vector_store", output_type=PGVector), + Output(display_name="Base Retriever", name="base_retriever", method="build_base_retriever", output_type=BaseRetriever), + Output(display_name="Search Results", name="search_results", method="search_documents"), + ] - def build_config(self): - """ - Builds the configuration for the component. + def build_vector_store(self) -> PGVector: + return self._build_pgvector() - Returns: - - dict: A dictionary containing the configuration options for the component. - """ - return { - "code": {"show": False}, - "inputs": {"display_name": "Input", "input_types": ["Document", "Data"]}, - "embedding": {"display_name": "Embedding"}, - "pg_server_url": { - "display_name": "PostgreSQL Server Connection String", - "advanced": False, - }, - "collection_name": {"display_name": "Table", "advanced": False}, - } + def build_base_retriever(self) -> BaseRetriever: + return self._build_pgvector() - def build( - self, - embedding: Embeddings, - pg_server_url: str, - collection_name: str, - inputs: Optional[Data] = None, - ) -> Union[VectorStore, BaseRetriever]: - """ - Builds the Vector Store or BaseRetriever object. + def _build_pgvector(self) -> PGVector: + if self.add_to_vector_store: + documents = [] + for _input in self.vector_store_inputs or []: + if isinstance(_input, Data): + documents.append(_input.to_lc_document()) + else: + documents.append(_input) - Args: - - embedding (Embeddings): The embeddings to use for the Vector Store. - - documents (Optional[Document]): The documents to use for the Vector Store. - - collection_name (str): The name of the PG table. - - pg_server_url (str): The URL for the PG server. - - Returns: - - VectorStore: The Vector Store object. - """ - - documents = [] - for _input in inputs or []: - if isinstance(_input, Data): - documents.append(_input.to_lc_document()) - else: - documents.append(_input) - try: - if documents is None: - vector_store = PGVector.from_existing_index( - embedding=embedding, - collection_name=collection_name, - connection_string=pg_server_url, + if documents: + pgvector = PGVector.from_documents( + embedding=self.embedding, + documents=documents, + collection_name=self.collection_name, + connection_string=self.pg_server_url, ) else: - vector_store = PGVector.from_documents( - embedding=embedding, - documents=documents, # type: ignore - collection_name=collection_name, - connection_string=pg_server_url, + pgvector = PGVector.from_existing_index( + embedding=self.embedding, + collection_name=self.collection_name, + connection_string=self.pg_server_url, ) - except Exception as e: - raise RuntimeError(f"Failed to build PGVector: {e}") - return vector_store + else: + pgvector = PGVector.from_existing_index( + embedding=self.embedding, + collection_name=self.collection_name, + connection_string=self.pg_server_url, + ) + + return pgvector + + def search_documents(self) -> List[Data]: + vector_store = self._build_pgvector() + + if self.search_input and isinstance(self.search_input, str) and self.search_input.strip(): + docs = vector_store.similarity_search( + query=self.search_input, + k=self.number_of_results, + ) + + data = docs_to_data(docs) + self.status = data + return data + else: + return []