""" NOTE: This vector database integration is community-supported and maintained on a best-effort basis. """ import logging from typing import Any, Dict, List, Optional, Union import boto3 from open_webui.config import S3_VECTOR_BUCKET_NAME, S3_VECTOR_REGION from open_webui.retrieval.vector.main import ( GetResult, SearchResult, VectorDBBase, VectorItem, ) from open_webui.retrieval.vector.utils import process_metadata log = logging.getLogger(__name__) class S3VectorClient(VectorDBBase): """ AWS S3 Vector integration for Open WebUI Knowledge. """ def __init__(self): self.bucket_name = S3_VECTOR_BUCKET_NAME self.region = S3_VECTOR_REGION # Simple validation - log warnings instead of raising exceptions if not self.bucket_name: log.warning('S3_VECTOR_BUCKET_NAME not set - S3Vector will not work') if not self.region: log.warning('S3_VECTOR_REGION not set - S3Vector will not work') if self.bucket_name and self.region: try: self.client = boto3.client('s3vectors', region_name=self.region) log.info("S3Vector client initialized for bucket '%s' in region '%s'", self.bucket_name, self.region) except Exception as e: log.error(f'Failed to initialize S3Vector client: {e}') self.client = None else: self.client = None def _create_index( self, index_name: str, dimension: int, data_type: str = 'float32', distance_metric: str = 'cosine', ) -> None: """ Create a new index in the S3 vector bucket for the given collection if it does not exist. """ if self.has_collection(index_name): log.debug("Index '%s' already exists, skipping creation", index_name) return try: self.client.create_index( vectorBucketName=self.bucket_name, indexName=index_name, dataType=data_type, dimension=dimension, distanceMetric=distance_metric, metadataConfiguration={ 'nonFilterableMetadataKeys': [ 'text', ] }, ) log.info( 'Created S3 index: %s (dim=%s, type=%s, metric=%s)', index_name, dimension, data_type, distance_metric ) except Exception as e: log.error(f"Error creating S3 index '{index_name}': {e}") raise def _filter_metadata(self, metadata: Dict[str, Any], item_id: str) -> Dict[str, Any]: """ Filter vector metadata keys to comply with S3 Vector API limit of 10 keys maximum. """ if not isinstance(metadata, dict) or len(metadata) <= 10: return metadata # Keep only the first 10 keys, prioritizing important ones based on actual Open WebUI metadata important_keys = [ 'text', # The actual document content 'file_id', # File ID 'source', # Document source file 'title', # Document title 'page', # Page number 'total_pages', # Total pages in document 'embedding_config', # Embedding configuration 'created_by', # User who created it 'name', # Document name 'hash', # Content hash ] filtered_metadata = {} # First, add important keys if they exist for key in important_keys: if key in metadata: filtered_metadata[key] = metadata[key] if len(filtered_metadata) >= 10: break # If we still have room, add other keys if len(filtered_metadata) < 10: for key, value in metadata.items(): if key not in filtered_metadata: filtered_metadata[key] = value if len(filtered_metadata) >= 10: break log.warning(f"Metadata for key '{item_id}' had {len(metadata)} keys, limited to 10 keys") return filtered_metadata def has_collection(self, collection_name: str) -> bool: """ Check if a vector index exists using direct lookup. This avoids pagination issues with list_indexes() and is significantly faster. """ try: self.client.get_index(vectorBucketName=self.bucket_name, indexName=collection_name) return True except Exception as e: log.error(f"Error checking if index '{collection_name}' exists: {e}") return False def delete_collection(self, collection_name: str) -> None: """ Delete an entire S3 Vector index/collection. """ if not self.has_collection(collection_name): log.warning(f"Collection '{collection_name}' does not exist, nothing to delete") return try: log.info("Deleting collection '%s'", collection_name) self.client.delete_index(vectorBucketName=self.bucket_name, indexName=collection_name) log.info("Successfully deleted collection '%s'", collection_name) except Exception as e: log.error(f"Error deleting collection '{collection_name}': {e}") raise def insert(self, collection_name: str, items: List[VectorItem]) -> None: """ Insert vector items into the S3 Vector index. Create index if it does not exist. """ if not items: log.warning('No items to insert') return dimension = len(items[0]['vector']) try: if not self.has_collection(collection_name): log.info("Index '%s' does not exist. Creating index.", collection_name) self._create_index( index_name=collection_name, dimension=dimension, data_type='float32', distance_metric='cosine', ) # Prepare vectors for insertion vectors = [] for item in items: # Ensure vector data is in the correct format for S3 Vector API vector_data = item['vector'] if isinstance(vector_data, list): # Convert list to float32 values as required by S3 Vector API vector_data = [float(x) for x in vector_data] # Prepare metadata, ensuring the text field is preserved metadata = item.get('metadata', {}).copy() # Add the text field to metadata so it's available for retrieval metadata['text'] = item['text'] # Convert metadata to string format for consistency metadata = process_metadata(metadata) # Filter metadata to comply with S3 Vector API limit of 10 keys metadata = self._filter_metadata(metadata, item['id']) vectors.append( { 'key': item['id'], 'data': {'float32': vector_data}, 'metadata': metadata, } ) # Insert vectors in batches of 500 (S3 Vector API limit) batch_size = 500 for i in range(0, len(vectors), batch_size): batch = vectors[i : i + batch_size] self.client.put_vectors( vectorBucketName=self.bucket_name, indexName=collection_name, vectors=batch, ) log.info( "Inserted batch %s: %s vectors into index '%s'.", i // batch_size + 1, len(batch), collection_name ) log.info("Completed insertion of %s vectors into index '%s'.", len(vectors), collection_name) except Exception as e: log.error(f'Error inserting vectors: {e}') raise def upsert(self, collection_name: str, items: List[VectorItem]) -> None: """ Insert or update vector items in the S3 Vector index. Create index if it does not exist. """ if not items: log.warning('No items to upsert') return dimension = len(items[0]['vector']) log.info('Upsert dimension: %s', dimension) try: if not self.has_collection(collection_name): log.info("Index '%s' does not exist. Creating index for upsert.", collection_name) self._create_index( index_name=collection_name, dimension=dimension, data_type='float32', distance_metric='cosine', ) # Prepare vectors for upsert vectors = [] for item in items: # Ensure vector data is in the correct format for S3 Vector API vector_data = item['vector'] if isinstance(vector_data, list): # Convert list to float32 values as required by S3 Vector API vector_data = [float(x) for x in vector_data] # Prepare metadata, ensuring the text field is preserved metadata = item.get('metadata', {}).copy() # Add the text field to metadata so it's available for retrieval metadata['text'] = item['text'] # Convert metadata to string format for consistency metadata = process_metadata(metadata) # Filter metadata to comply with S3 Vector API limit of 10 keys metadata = self._filter_metadata(metadata, item['id']) vectors.append( { 'key': item['id'], 'data': {'float32': vector_data}, 'metadata': metadata, } ) # Upsert vectors in batches of 500 (S3 Vector API limit) batch_size = 500 for i in range(0, len(vectors), batch_size): batch = vectors[i : i + batch_size] if i == 0: # Log sample info for first batch only log.info( 'Upserting batch 1: %s vectors. First vector sample: key=%s, data_type=%s, data_len=%s', len(batch), batch[0]['key'], type(batch[0]['data']['float32']), len(batch[0]['data']['float32']), ) else: log.info('Upserting batch %s: %s vectors.', i // batch_size + 1, len(batch)) self.client.put_vectors( vectorBucketName=self.bucket_name, indexName=collection_name, vectors=batch, ) log.info("Completed upsert of %s vectors into index '%s'.", len(vectors), collection_name) except Exception as e: log.error(f'Error upserting vectors: {e}') raise def search( self, collection_name: str, vectors: List[List[Union[float, int]]], filter: Optional[dict] = None, limit: int = 10, ) -> Optional[SearchResult]: """ Search for similar vectors in a collection using multiple query vectors. """ if not self.has_collection(collection_name): log.warning(f"Collection '{collection_name}' does not exist") return None if not vectors: log.warning('No query vectors provided') return None try: log.info("Searching collection '%s' with %s query vectors, limit=%s", collection_name, len(vectors), limit) # Initialize result lists all_ids = [] all_documents = [] all_metadatas = [] all_distances = [] # Process each query vector for i, query_vector in enumerate(vectors): log.debug('Processing query vector %s/%s', i + 1, len(vectors)) # Prepare the query vector in S3 Vector format query_vector_dict = {'float32': [float(x) for x in query_vector]} # Call S3 Vector query API response = self.client.query_vectors( vectorBucketName=self.bucket_name, indexName=collection_name, topK=limit, queryVector=query_vector_dict, returnMetadata=True, returnDistance=True, ) # Process results for this query query_ids = [] query_documents = [] query_metadatas = [] query_distances = [] result_vectors = response.get('vectors', []) for vector in result_vectors: vector_id = vector.get('key') vector_metadata = vector.get('metadata', {}) vector_distance = vector.get('distance', 0.0) # Extract document text from metadata document_text = '' if isinstance(vector_metadata, dict): # Get the text field first (highest priority) document_text = vector_metadata.get('text') if not document_text: # Fallback to other possible text fields document_text = ( vector_metadata.get('content') or vector_metadata.get('document') or vector_id ) else: document_text = vector_id query_ids.append(vector_id) query_documents.append(document_text) query_metadatas.append(vector_metadata) query_distances.append(vector_distance) # Add this query's results to the overall results all_ids.append(query_ids) all_documents.append(query_documents) all_metadatas.append(query_metadatas) all_distances.append(query_distances) log.info('Search completed. Found results for %s queries', len(all_ids)) # Return SearchResult format return SearchResult( ids=all_ids if all_ids else None, documents=all_documents if all_documents else None, metadatas=all_metadatas if all_metadatas else None, distances=all_distances if all_distances else None, ) except Exception as e: log.error(f"Error searching collection '{collection_name}': {str(e)}") # Handle specific AWS exceptions if hasattr(e, 'response') and 'Error' in e.response: error_code = e.response['Error']['Code'] if error_code == 'NotFoundException': log.warning(f"Collection '{collection_name}' not found") return None elif error_code == 'ValidationException': log.error(f'Invalid query vector dimensions or parameters') return None elif error_code == 'AccessDeniedException': log.error(f"Access denied for collection '{collection_name}'. Check permissions.") return None raise def query(self, collection_name: str, filter: Dict, limit: Optional[int] = None) -> Optional[GetResult]: """ Query vectors from a collection using metadata filter. """ if not self.has_collection(collection_name): log.warning(f"Collection '{collection_name}' does not exist") return GetResult(ids=[[]], documents=[[]], metadatas=[[]]) if not filter: log.warning('No filter provided, returning all vectors') return self.get(collection_name) try: log.info("Querying collection '%s' with filter: %s", collection_name, filter) # For S3 Vector, we need to use list_vectors and then filter results # Since S3 Vector may not support complex server-side filtering, # we'll retrieve all vectors and filter client-side # Get all vectors first all_vectors_result = self.get(collection_name) if not all_vectors_result or not all_vectors_result.ids: log.warning('No vectors found in collection') return GetResult(ids=[[]], documents=[[]], metadatas=[[]]) # Extract the lists from the result all_ids = all_vectors_result.ids[0] if all_vectors_result.ids else [] all_documents = all_vectors_result.documents[0] if all_vectors_result.documents else [] all_metadatas = all_vectors_result.metadatas[0] if all_vectors_result.metadatas else [] # Apply client-side filtering filtered_ids = [] filtered_documents = [] filtered_metadatas = [] for i, metadata in enumerate(all_metadatas): if self._matches_filter(metadata, filter): if i < len(all_ids): filtered_ids.append(all_ids[i]) if i < len(all_documents): filtered_documents.append(all_documents[i]) filtered_metadatas.append(metadata) # Apply limit if specified if limit and len(filtered_ids) >= limit: break log.info('Filter applied: %s vectors match out of %s total', len(filtered_ids), len(all_ids)) # Return GetResult format if filtered_ids: return GetResult( ids=[filtered_ids], documents=[filtered_documents], metadatas=[filtered_metadatas], ) else: return GetResult(ids=[[]], documents=[[]], metadatas=[[]]) except Exception as e: log.error(f"Error querying collection '{collection_name}': {str(e)}") # Handle specific AWS exceptions if hasattr(e, 'response') and 'Error' in e.response: error_code = e.response['Error']['Code'] if error_code == 'NotFoundException': log.warning(f"Collection '{collection_name}' not found") return GetResult(ids=[[]], documents=[[]], metadatas=[[]]) elif error_code == 'AccessDeniedException': log.error(f"Access denied for collection '{collection_name}'. Check permissions.") return GetResult(ids=[[]], documents=[[]], metadatas=[[]]) raise def get(self, collection_name: str) -> Optional[GetResult]: """ Retrieve all vectors from a collection. """ if not self.has_collection(collection_name): log.warning(f"Collection '{collection_name}' does not exist") return GetResult(ids=[[]], documents=[[]], metadatas=[[]]) try: log.info("Retrieving all vectors from collection '%s'", collection_name) # Initialize result lists all_ids = [] all_documents = [] all_metadatas = [] # Handle pagination next_token = None while True: # Prepare request parameters request_params = { 'vectorBucketName': self.bucket_name, 'indexName': collection_name, 'returnData': False, # Don't include vector data (not needed for get) 'returnMetadata': True, # Include metadata 'maxResults': 500, # Use reasonable page size } if next_token: request_params['nextToken'] = next_token # Call S3 Vector API response = self.client.list_vectors(**request_params) # Process vectors in this page vectors = response.get('vectors', []) for vector in vectors: vector_id = vector.get('key') vector_data = vector.get('data', {}) vector_metadata = vector.get('metadata', {}) # Extract the actual vector array vector_array = vector_data.get('float32', []) # For documents, we try to extract text from metadata or use the vector ID document_text = '' if isinstance(vector_metadata, dict): # Get the text field first (highest priority) document_text = vector_metadata.get('text') if not document_text: # Fallback to other possible text fields document_text = ( vector_metadata.get('content') or vector_metadata.get('document') or vector_id ) # Log the actual content for debugging log.debug('Document text preview (first 200 chars): %s', str(document_text)[:200]) else: document_text = vector_id all_ids.append(vector_id) all_documents.append(document_text) all_metadatas.append(vector_metadata) # Check if there are more pages next_token = response.get('nextToken') if not next_token: break log.info("Retrieved %s vectors from collection '%s'", len(all_ids), collection_name) # Return in GetResult format # The Open WebUI GetResult expects lists of lists, so we wrap each list if all_ids: return GetResult(ids=[all_ids], documents=[all_documents], metadatas=[all_metadatas]) else: return GetResult(ids=[[]], documents=[[]], metadatas=[[]]) except Exception as e: log.error(f"Error retrieving vectors from collection '{collection_name}': {str(e)}") # Handle specific AWS exceptions if hasattr(e, 'response') and 'Error' in e.response: error_code = e.response['Error']['Code'] if error_code == 'NotFoundException': log.warning(f"Collection '{collection_name}' not found") return GetResult(ids=[[]], documents=[[]], metadatas=[[]]) elif error_code == 'AccessDeniedException': log.error(f"Access denied for collection '{collection_name}'. Check permissions.") return GetResult(ids=[[]], documents=[[]], metadatas=[[]]) raise def delete( self, collection_name: str, ids: Optional[List[str]] = None, filter: Optional[Dict] = None, ) -> None: """ Delete vectors by ID or filter from a collection. """ if not self.has_collection(collection_name): log.warning(f"Collection '{collection_name}' does not exist, nothing to delete") return # Check if this is a knowledge collection (not file-specific) is_knowledge_collection = not collection_name.startswith('file-') try: if ids: # Delete by specific vector IDs/keys log.info("Deleting %s vectors by IDs from collection '%s'", len(ids), collection_name) self.client.delete_vectors( vectorBucketName=self.bucket_name, indexName=collection_name, keys=ids, ) log.info("Deleted %s vectors from index '%s'", len(ids), collection_name) elif filter: # Handle filter-based deletion log.info("Deleting vectors by filter from collection '%s': %s", collection_name, filter) # If this is a knowledge collection and we have a file_id filter, # also clean up the corresponding file-specific collection if is_knowledge_collection and 'file_id' in filter: file_id = filter['file_id'] file_collection_name = f'file-{file_id}' if self.has_collection(file_collection_name): log.info( "Found related file-specific collection '%s', deleting it to prevent duplicates", file_collection_name, ) self.delete_collection(file_collection_name) # For the main collection, implement query-then-delete # First, query to get IDs matching the filter query_result = self.query(collection_name, filter) if query_result and query_result.ids and query_result.ids[0]: matching_ids = query_result.ids[0] log.info('Found %s vectors matching filter, deleting them', len(matching_ids)) # Delete the matching vectors by ID self.client.delete_vectors( vectorBucketName=self.bucket_name, indexName=collection_name, keys=matching_ids, ) log.info("Deleted %s vectors from index '%s' using filter", len(matching_ids), collection_name) else: log.warning('No vectors found matching the filter criteria') else: log.warning('No IDs or filter provided for deletion') except Exception as e: log.error(f"Error deleting vectors from collection '{collection_name}': {e}") raise def reset(self) -> None: """ Reset/clear all vector data. For S3 Vector, this deletes all indexes. """ try: log.warning('Reset called - this will delete all vector indexes in the S3 bucket') # List all indexes response = self.client.list_indexes(vectorBucketName=self.bucket_name) indexes = response.get('indexes', []) if not indexes: log.warning('No indexes found to delete') return # Delete all indexes deleted_count = 0 for index in indexes: index_name = index.get('indexName') if index_name: try: self.client.delete_index(vectorBucketName=self.bucket_name, indexName=index_name) deleted_count += 1 log.info('Deleted index: %s', index_name) except Exception as e: log.error(f"Error deleting index '{index_name}': {e}") log.info('Reset completed: deleted %s indexes', deleted_count) except Exception as e: log.error(f'Error during reset: {e}') raise def _matches_filter(self, metadata: Dict[str, Any], filter: Dict[str, Any]) -> bool: """ Check if metadata matches the given filter conditions. """ if not isinstance(metadata, dict) or not isinstance(filter, dict): return False # Check each filter condition for key, expected_value in filter.items(): # Handle special operators if key.startswith('$'): if key == '$and': # All conditions must match if not isinstance(expected_value, list): continue for condition in expected_value: if not self._matches_filter(metadata, condition): return False elif key == '$or': # At least one condition must match if not isinstance(expected_value, list): continue any_match = False for condition in expected_value: if self._matches_filter(metadata, condition): any_match = True break if not any_match: return False continue # Get the actual value from metadata actual_value = metadata.get(key) # Handle different types of expected values if isinstance(expected_value, dict): # Handle comparison operators for op, op_value in expected_value.items(): if op == '$eq': if actual_value != op_value: return False elif op == '$ne': if actual_value == op_value: return False elif op == '$in': if not isinstance(op_value, list) or actual_value not in op_value: return False elif op == '$nin': if isinstance(op_value, list) and actual_value in op_value: return False elif op == '$exists': if bool(op_value) != (key in metadata): return False # Add more operators as needed else: # Simple equality check if actual_value != expected_value: return False return True