perf: build info log messages lazily so raising the log level actually saves work (#27837)
Raising GLOBAL_LOG_LEVEL to WARNING buys quieter output but not less work: 241 INFO call sites interpolate their payload into an f-string before the logging call gets to drop it. The heaviest is get_doc, which logs every chunk id and metadata dict in a collection, so on the full-context retrieval path that is the entire knowledge base, once per chat request.
That one line at WARNING, CPython 3.12:
| knowledge base | payload | before | after |
| -------------- | ------- | -------- | ------- |
| top-k of 3 | 1.2 kB | 3.8 us | 0.07 us |
| 500 chunks | 201 kB | 583.6 us | 0.08 us |
| 5000 chunks | 2.0 MB | 5.8 ms | 0.15 us |
The lazy form log.info('query_doc:result %s %s', result.ids, result.metadatas) hands the payload to record.getMessage(), which the InterceptHandler only reaches once a record has passed the level check. Output at INFO is byte-identical. Two sites that already built their message eagerly, one str concat and one % operator, move to the same lazy form.
This commit is contained in:
1 parent
52145eede9
commit
2d18727ab8
48 files changed
+297
-247
No files matched your search
@@ -72,7 +72,7 @@ class DatalabMarkerLoader:
|
||||
response = requests.get(url, headers=headers)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
log.info(f'Marker API status check for request {request_id}: {result}')
|
||||
log.info('Marker API status check for request %s: %s', request_id, result)
|
||||
return result
|
||||
except requests.HTTPError as e:
|
||||
log.error(f'Error checking Marker request status: {e}')
|
||||
@@ -104,7 +104,10 @@ class DatalabMarkerLoader:
|
||||
form_data['additional_config'] = self.additional_config
|
||||
|
||||
log.info(
|
||||
f"Datalab Marker POST request parameters: {{'filename': '{filename}', 'mime_type': '{mime_type}', **{form_data}}}"
|
||||
"Datalab Marker POST request parameters: {'filename': '%s', 'mime_type': '%s', **%s}",
|
||||
filename,
|
||||
mime_type,
|
||||
form_data,
|
||||
)
|
||||
|
||||
try:
|
||||
@@ -168,7 +171,7 @@ class DatalabMarkerLoader:
|
||||
'total_cost',
|
||||
)
|
||||
}
|
||||
log.info(f'Marker processing completed successfully: {json.dumps(summary, indent=2)}')
|
||||
log.info('Marker processing completed successfully: %s', json.dumps(summary, indent=2))
|
||||
break
|
||||
|
||||
if status_val == 'failed' or success_val is False:
|
||||
@@ -235,7 +238,7 @@ class DatalabMarkerLoader:
|
||||
try:
|
||||
with open(output_path, 'w', encoding='utf-8') as f:
|
||||
f.write(full_text)
|
||||
log.info(f'Saved Marker output to: {output_path}')
|
||||
log.info('Saved Marker output to: %s', output_path)
|
||||
except Exception as e:
|
||||
log.warning(f'Failed to write marker output to disk: {e}')
|
||||
|
||||
|
||||
@@ -74,7 +74,7 @@ class MinerULoader:
|
||||
Load document using Local API (synchronous).
|
||||
Posts file to /file_parse endpoint and gets immediate response.
|
||||
"""
|
||||
log.info(f'Using MinerU Local API at {self.api_url}')
|
||||
log.info('Using MinerU Local API at %s', self.api_url)
|
||||
|
||||
filename = os.path.basename(self.file_path)
|
||||
|
||||
@@ -97,7 +97,7 @@ class MinerULoader:
|
||||
with open(self.file_path, 'rb') as f:
|
||||
files = {'files': (filename, f, 'application/octet-stream')}
|
||||
|
||||
log.info(f'Sending file to MinerU Local API: {filename}')
|
||||
log.info('Sending file to MinerU Local API: %s', filename)
|
||||
log.debug('Local API parameters: %s', form_data)
|
||||
|
||||
response = requests.post(
|
||||
@@ -163,7 +163,7 @@ class MinerULoader:
|
||||
detail='MinerU returned empty markdown content',
|
||||
)
|
||||
|
||||
log.info(f'Successfully parsed document with MinerU Local API: {filename}')
|
||||
log.info('Successfully parsed document with MinerU Local API: %s', filename)
|
||||
|
||||
# Create metadata
|
||||
metadata = {
|
||||
@@ -180,7 +180,7 @@ class MinerULoader:
|
||||
Load document using Cloud API (asynchronous).
|
||||
Uses batch upload endpoint to avoid need for public file URLs.
|
||||
"""
|
||||
log.info(f'Using MinerU Cloud API at {self.api_url}')
|
||||
log.info('Using MinerU Cloud API at %s', self.api_url)
|
||||
|
||||
filename = os.path.basename(self.file_path)
|
||||
|
||||
@@ -196,7 +196,7 @@ class MinerULoader:
|
||||
# Step 4: Download and extract markdown from ZIP
|
||||
markdown_content = self._download_and_extract_zip(result['full_zip_url'], filename)
|
||||
|
||||
log.info(f'Successfully parsed document with MinerU Cloud API: {filename}')
|
||||
log.info('Successfully parsed document with MinerU Cloud API: %s', filename)
|
||||
|
||||
# Create metadata
|
||||
metadata = {
|
||||
@@ -232,7 +232,7 @@ class MinerULoader:
|
||||
if self.page_ranges:
|
||||
request_body['files'][0]['page_ranges'] = self.page_ranges
|
||||
|
||||
log.info(f'Requesting upload URL for: {filename}')
|
||||
log.info('Requesting upload URL for: %s', filename)
|
||||
log.debug('Cloud API request body: %s', request_body)
|
||||
|
||||
try:
|
||||
@@ -284,7 +284,7 @@ class MinerULoader:
|
||||
)
|
||||
|
||||
upload_url = file_urls[0]
|
||||
log.info(f'Received upload URL for batch: {batch_id}')
|
||||
log.info('Received upload URL for batch: %s', batch_id)
|
||||
|
||||
return batch_id, upload_url
|
||||
|
||||
@@ -334,7 +334,7 @@ class MinerULoader:
|
||||
max_iterations = 300 # 10 minutes max (2 seconds per iteration)
|
||||
poll_interval = 2 # seconds
|
||||
|
||||
log.info(f'Polling batch status: {batch_id}')
|
||||
log.info('Polling batch status: %s', batch_id)
|
||||
|
||||
for iteration in range(max_iterations):
|
||||
try:
|
||||
@@ -393,7 +393,7 @@ class MinerULoader:
|
||||
state = file_result.get('state')
|
||||
|
||||
if state == 'done':
|
||||
log.info(f'Processing complete for {filename}')
|
||||
log.info('Processing complete for %s', filename)
|
||||
return file_result
|
||||
elif state == 'failed':
|
||||
error_msg = file_result.get('err_msg', 'Unknown error')
|
||||
@@ -404,7 +404,7 @@ class MinerULoader:
|
||||
elif state in ['waiting-file', 'pending', 'running', 'converting']:
|
||||
# Still processing
|
||||
if iteration % 10 == 0: # Log every 20 seconds
|
||||
log.info(f'Processing status: {state} (iteration {iteration + 1}/{max_iterations})')
|
||||
log.info('Processing status: %s (iteration %s/%s)', state, iteration + 1, max_iterations)
|
||||
time.sleep(poll_interval)
|
||||
else:
|
||||
log.warning(f'Unknown state: {state}')
|
||||
@@ -421,7 +421,7 @@ class MinerULoader:
|
||||
Download ZIP file from CDN and extract markdown content.
|
||||
Returns the markdown content as a string.
|
||||
"""
|
||||
log.info(f'Downloading results from: {zip_url}')
|
||||
log.info('Downloading results from: %s', zip_url)
|
||||
|
||||
try:
|
||||
response = requests.get(zip_url, timeout=60)
|
||||
@@ -452,7 +452,7 @@ class MinerULoader:
|
||||
read_errors = []
|
||||
|
||||
for member in md_members:
|
||||
log.info(f'Found markdown file in ZIP: {member.filename}')
|
||||
log.info('Found markdown file in ZIP: %s', member.filename)
|
||||
try:
|
||||
with zip_ref.open(member, 'r') as f:
|
||||
if self.max_markdown_bytes is None:
|
||||
@@ -515,5 +515,5 @@ class MinerULoader:
|
||||
detail='Extracted markdown content is empty',
|
||||
)
|
||||
|
||||
log.info(f'Successfully extracted markdown content ({len(markdown_content)} characters)')
|
||||
log.info('Successfully extracted markdown content (%s characters)', len(markdown_content))
|
||||
return markdown_content
|
||||
@@ -261,7 +261,7 @@ class MistralLoader:
|
||||
file_id = response_data.get('id')
|
||||
if not file_id:
|
||||
raise ValueError('File ID not found in upload response.')
|
||||
log.info(f'File uploaded successfully. File ID: {file_id}')
|
||||
log.info('File uploaded successfully. File ID: %s', file_id)
|
||||
return file_id
|
||||
except Exception as e:
|
||||
log.error(f'Failed to upload file: {e}')
|
||||
@@ -305,12 +305,12 @@ class MistralLoader:
|
||||
if not file_id:
|
||||
raise ValueError('File ID not found in upload response.')
|
||||
|
||||
log.info(f'File uploaded successfully. File ID: {file_id}')
|
||||
log.info('File uploaded successfully. File ID: %s', file_id)
|
||||
return file_id
|
||||
|
||||
def _get_signed_url(self, file_id: str) -> str:
|
||||
"""Retrieves a temporary signed URL for the uploaded file (sync version)."""
|
||||
log.info(f'Getting signed URL for file ID: {file_id}')
|
||||
log.info('Getting signed URL for file ID: %s', file_id)
|
||||
url = f'{self.base_url}/files/{file_id}/url'
|
||||
params = {'expiry': 1}
|
||||
signed_url_headers = {**self.headers, 'Accept': 'application/json'}
|
||||
@@ -421,7 +421,7 @@ class MistralLoader:
|
||||
ocr_response = await self._handle_response_async(response)
|
||||
|
||||
processing_time = time.time() - start_time
|
||||
log.info(f'OCR processing completed in {processing_time:.2f}s')
|
||||
log.info('OCR processing completed in %.2fs', processing_time)
|
||||
|
||||
return ocr_response
|
||||
|
||||
@@ -434,13 +434,13 @@ class MistralLoader:
|
||||
|
||||
def _delete_file(self, file_id: str) -> None:
|
||||
"""Deletes the file from Mistral storage (sync version)."""
|
||||
log.info(f'Deleting uploaded file ID: {file_id}')
|
||||
log.info('Deleting uploaded file ID: %s', file_id)
|
||||
url = f'{self.base_url}/files/{file_id}'
|
||||
|
||||
try:
|
||||
response = requests.delete(url, headers=self.headers, timeout=self.cleanup_timeout)
|
||||
delete_response = self._handle_response(response)
|
||||
log.info(f'File deleted successfully: {delete_response}')
|
||||
log.info('File deleted successfully: %s', delete_response)
|
||||
except Exception as e:
|
||||
# Log error but don't necessarily halt execution if deletion fails
|
||||
log.error(f'Failed to delete file ID {file_id}: {e}')
|
||||
@@ -551,7 +551,7 @@ class MistralLoader:
|
||||
)
|
||||
|
||||
if skipped_pages > 0:
|
||||
log.info(f'Processed {len(documents)} pages, skipped {skipped_pages} empty/invalid pages')
|
||||
log.info('Processed %s pages, skipped %s empty/invalid pages', len(documents), skipped_pages)
|
||||
|
||||
if not documents:
|
||||
# Case where pages existed but none had valid markdown/index
|
||||
@@ -584,7 +584,7 @@ class MistralLoader:
|
||||
if self.use_base64:
|
||||
documents = self._process_results(self._process_ocr(self._get_file_data_url()))
|
||||
total_time = time.time() - start_time
|
||||
log.info(f'Sync OCR workflow completed in {total_time:.2f}s, produced {len(documents)} documents')
|
||||
log.info('Sync OCR workflow completed in %.2fs, produced %s documents', total_time, len(documents))
|
||||
return documents
|
||||
|
||||
# 1. Upload file
|
||||
@@ -600,7 +600,7 @@ class MistralLoader:
|
||||
documents = self._process_results(ocr_response)
|
||||
|
||||
total_time = time.time() - start_time
|
||||
log.info(f'Sync OCR workflow completed in {total_time:.2f}s, produced {len(documents)} documents')
|
||||
log.info('Sync OCR workflow completed in %.2fs, produced %s documents', total_time, len(documents))
|
||||
|
||||
return documents
|
||||
|
||||
@@ -642,7 +642,7 @@ class MistralLoader:
|
||||
ocr_response = await self._process_ocr_async(session, self._get_file_data_url())
|
||||
documents = self._process_results(ocr_response)
|
||||
total_time = time.time() - start_time
|
||||
log.info(f'Async OCR workflow completed in {total_time:.2f}s, produced {len(documents)} documents')
|
||||
log.info('Async OCR workflow completed in %.2fs, produced %s documents', total_time, len(documents))
|
||||
return documents
|
||||
|
||||
# 1. Upload file with streaming
|
||||
@@ -658,7 +658,7 @@ class MistralLoader:
|
||||
documents = self._process_results(ocr_response)
|
||||
|
||||
total_time = time.time() - start_time
|
||||
log.info(f'Async OCR workflow completed in {total_time:.2f}s, produced {len(documents)} documents')
|
||||
log.info('Async OCR workflow completed in %.2fs, produced %s documents', total_time, len(documents))
|
||||
|
||||
return documents
|
||||
|
||||
@@ -701,7 +701,7 @@ class MistralLoader:
|
||||
if not loaders:
|
||||
return []
|
||||
|
||||
log.info(f'Starting concurrent processing of {len(loaders)} files with max {max_concurrent} concurrent')
|
||||
log.info('Starting concurrent processing of %s files with max %s concurrent', len(loaders), max_concurrent)
|
||||
start_time = time.time()
|
||||
|
||||
# Use semaphore to control concurrency
|
||||
@@ -741,9 +741,11 @@ class MistralLoader:
|
||||
failure_count = len(results) - success_count
|
||||
|
||||
log.info(
|
||||
f'Batch processing completed in {total_time:.2f}s: '
|
||||
f'{success_count} files succeeded, {failure_count} files failed, '
|
||||
f'produced {total_docs} total documents'
|
||||
'Batch processing completed in %.2fs: %s files succeeded, %s files failed, produced %s total documents',
|
||||
total_time,
|
||||
success_count,
|
||||
failure_count,
|
||||
total_docs,
|
||||
)
|
||||
|
||||
return processed_results
|
||||
@@ -35,7 +35,7 @@ class PaddleOCRVLLoader:
|
||||
self.file_name = os.path.basename(file_path)
|
||||
|
||||
def load(self) -> List[Document]:
|
||||
log.info(f'Processing with PaddleOCR-vl: {self.file_path}')
|
||||
log.info('Processing with PaddleOCR-vl: %s', self.file_path)
|
||||
|
||||
try:
|
||||
with open(self.file_path, 'rb') as file:
|
||||
@@ -96,7 +96,7 @@ class PaddleOCRVLLoader:
|
||||
)
|
||||
|
||||
if skipped_pages > 0:
|
||||
log.info(f'PaddleOCR-vl: Processed {len(documents)} pages, skipped {skipped_pages} empty pages.')
|
||||
log.info('PaddleOCR-vl: Processed %s pages, skipped %s empty pages.', len(documents), skipped_pages)
|
||||
|
||||
if not documents:
|
||||
log.warning('No valid text content found by PaddleOCR-vl.')
|
||||
|
||||
@@ -138,7 +138,7 @@ class YoutubeLoader:
|
||||
log.debug("No transcript found for language '%s'", lang)
|
||||
continue
|
||||
except Exception as e:
|
||||
log.info(f"Error finding transcript for language '{lang}'")
|
||||
log.info("Error finding transcript for language '%s'", lang)
|
||||
raise e
|
||||
|
||||
# If we get here, all languages failed
|
||||
|
||||
@@ -35,8 +35,8 @@ class ExternalReranker(BaseReranker):
|
||||
}
|
||||
|
||||
try:
|
||||
log.info(f'ExternalReranker:predict:model {self.model}')
|
||||
log.info(f'ExternalReranker:predict:query {query}')
|
||||
log.info('ExternalReranker:predict:model %s', self.model)
|
||||
log.info('ExternalReranker:predict:query %s', query)
|
||||
|
||||
headers = {
|
||||
'Content-Type': 'application/json',
|
||||
|
||||
@@ -318,7 +318,7 @@ def query_doc(collection_name: str, query_embedding: list[float], k: int, user:
|
||||
)
|
||||
|
||||
if result:
|
||||
log.info(f'query_doc:result {result.ids} {result.metadatas}')
|
||||
log.info('query_doc:result %s %s', result.ids, result.metadatas)
|
||||
|
||||
return result
|
||||
except Exception as e:
|
||||
@@ -332,7 +332,7 @@ def get_doc(collection_name: str, user: UserModel = None):
|
||||
result = VECTOR_DB_CLIENT.get(collection_name=collection_name)
|
||||
|
||||
if result:
|
||||
log.info(f'query_doc:result {result.ids} {result.metadatas}')
|
||||
log.info('query_doc:result %s %s', result.ids, result.metadatas)
|
||||
|
||||
return result
|
||||
except Exception as e:
|
||||
@@ -585,7 +585,7 @@ async def query_doc_with_hybrid_search(
|
||||
'metadatas': [metadatas],
|
||||
}
|
||||
|
||||
log.info('query_doc_with_hybrid_search:result ' + f'{result["metadatas"]} {result["distances"]}')
|
||||
log.info('query_doc_with_hybrid_search:result %s %s', result['metadatas'], result['distances'])
|
||||
return result
|
||||
except Exception as e:
|
||||
log.exception(f'Error querying doc {collection_name} with hybrid search: {e}')
|
||||
@@ -812,7 +812,7 @@ async def query_collection_with_hybrid_search(
|
||||
|
||||
collection_results = dict(await asyncio.gather(*(_fetch_collection(name) for name in collection_names)))
|
||||
|
||||
log.info(f'Starting hybrid search for {len(queries)} queries in {len(collection_names)} collections...')
|
||||
log.info('Starting hybrid search for %s queries in %s collections...', len(queries), len(collection_names))
|
||||
|
||||
async def process_query(collection_name, query):
|
||||
try:
|
||||
|
||||
@@ -125,7 +125,7 @@ class MilvusClient(VectorDBBase):
|
||||
index_type = MILVUS_INDEX_TYPE.upper()
|
||||
metric_type = MILVUS_METRIC_TYPE.upper()
|
||||
|
||||
log.info(f'Using Milvus index type: {index_type}, metric type: {metric_type}')
|
||||
log.info('Using Milvus index type: %s, metric type: %s', index_type, metric_type)
|
||||
|
||||
index_creation_params = {}
|
||||
if index_type == 'HNSW':
|
||||
@@ -133,18 +133,18 @@ class MilvusClient(VectorDBBase):
|
||||
'M': MILVUS_HNSW_M,
|
||||
'efConstruction': MILVUS_HNSW_EFCONSTRUCTION,
|
||||
}
|
||||
log.info(f'HNSW params: {index_creation_params}')
|
||||
log.info('HNSW params: %s', index_creation_params)
|
||||
elif index_type == 'IVF_FLAT':
|
||||
index_creation_params = {'nlist': MILVUS_IVF_FLAT_NLIST}
|
||||
log.info(f'IVF_FLAT params: {index_creation_params}')
|
||||
log.info('IVF_FLAT params: %s', index_creation_params)
|
||||
elif index_type == 'DISKANN':
|
||||
index_creation_params = {
|
||||
'max_degree': MILVUS_DISKANN_MAX_DEGREE,
|
||||
'search_list_size': MILVUS_DISKANN_SEARCH_LIST_SIZE,
|
||||
}
|
||||
log.info(f'DISKANN params: {index_creation_params}')
|
||||
log.info('DISKANN params: %s', index_creation_params)
|
||||
elif index_type in ['FLAT', 'AUTOINDEX']:
|
||||
log.info(f'Using {index_type} index with no specific build-time params.')
|
||||
log.info('Using %s index with no specific build-time params.', index_type)
|
||||
else:
|
||||
log.warning(
|
||||
f"Unsupported MILVUS_INDEX_TYPE: '{index_type}'. "
|
||||
@@ -167,7 +167,11 @@ class MilvusClient(VectorDBBase):
|
||||
index_params=index_params,
|
||||
)
|
||||
log.info(
|
||||
f"Successfully created collection '{self.collection_prefix}_{collection_name}' with index type '{index_type}' and metric '{metric_type}'."
|
||||
"Successfully created collection '%s_%s' with index type '%s' and metric '%s'.",
|
||||
self.collection_prefix,
|
||||
collection_name,
|
||||
index_type,
|
||||
metric_type,
|
||||
)
|
||||
|
||||
def has_collection(self, collection_name: str) -> bool:
|
||||
@@ -220,7 +224,11 @@ class MilvusClient(VectorDBBase):
|
||||
|
||||
try:
|
||||
log.info(
|
||||
f"Querying collection {self.collection_prefix}_{collection_name} with filter: '{filter_string}', limit: {limit}"
|
||||
"Querying collection %s_%s with filter: '%s', limit: %s",
|
||||
self.collection_prefix,
|
||||
collection_name,
|
||||
filter_string,
|
||||
limit,
|
||||
)
|
||||
|
||||
iterator = self.client.query_iterator(
|
||||
@@ -265,7 +273,7 @@ class MilvusClient(VectorDBBase):
|
||||
# Insert the items into the collection, if the collection does not exist, it will be created.
|
||||
collection_name = collection_name.replace('-', '_')
|
||||
if not self.client.has_collection(collection_name=f'{self.collection_prefix}_{collection_name}'):
|
||||
log.info(f'Collection {self.collection_prefix}_{collection_name} does not exist. Creating now.')
|
||||
log.info('Collection %s_%s does not exist. Creating now.', self.collection_prefix, collection_name)
|
||||
if not items:
|
||||
log.error(
|
||||
f'Cannot create collection {self.collection_prefix}_{collection_name} without items to determine dimension.'
|
||||
@@ -273,7 +281,7 @@ class MilvusClient(VectorDBBase):
|
||||
raise ValueError('Cannot create Milvus collection without items to determine vector dimension.')
|
||||
self._create_collection(collection_name=collection_name, dimension=len(items[0]['vector']))
|
||||
|
||||
log.info(f'Inserting {len(items)} items into collection {self.collection_prefix}_{collection_name}.')
|
||||
log.info('Inserting %s items into collection %s_%s.', len(items), self.collection_prefix, collection_name)
|
||||
data = []
|
||||
for item in items:
|
||||
text = item['text'] or ''
|
||||
@@ -301,7 +309,9 @@ class MilvusClient(VectorDBBase):
|
||||
# Update the items in the collection, if the items are not present, insert them. If the collection does not exist, it will be created.
|
||||
collection_name = collection_name.replace('-', '_')
|
||||
if not self.client.has_collection(collection_name=f'{self.collection_prefix}_{collection_name}'):
|
||||
log.info(f'Collection {self.collection_prefix}_{collection_name} does not exist for upsert. Creating now.')
|
||||
log.info(
|
||||
'Collection %s_%s does not exist for upsert. Creating now.', self.collection_prefix, collection_name
|
||||
)
|
||||
if not items:
|
||||
log.error(
|
||||
f'Cannot create collection {self.collection_prefix}_{collection_name} for upsert without items to determine dimension.'
|
||||
@@ -311,7 +321,7 @@ class MilvusClient(VectorDBBase):
|
||||
)
|
||||
self._create_collection(collection_name=collection_name, dimension=len(items[0]['vector']))
|
||||
|
||||
log.info(f'Upserting {len(items)} items into collection {self.collection_prefix}_{collection_name}.')
|
||||
log.info('Upserting %s items into collection %s_%s.', len(items), self.collection_prefix, collection_name)
|
||||
data = []
|
||||
for item in items:
|
||||
text = item['text'] or ''
|
||||
@@ -348,7 +358,7 @@ class MilvusClient(VectorDBBase):
|
||||
return None
|
||||
|
||||
if ids:
|
||||
log.info(f'Deleting items by IDs from {self.collection_prefix}_{collection_name}. IDs: {ids}')
|
||||
log.info('Deleting items by IDs from %s_%s. IDs: %s', self.collection_prefix, collection_name, ids)
|
||||
return self.client.delete(
|
||||
collection_name=f'{self.collection_prefix}_{collection_name}',
|
||||
ids=ids,
|
||||
@@ -356,7 +366,10 @@ class MilvusClient(VectorDBBase):
|
||||
elif filter:
|
||||
filter_string = ' && '.join([f'metadata["{key}"] == {JSONCodec.dumps(value)}' for key, value in filter.items()])
|
||||
log.info(
|
||||
f'Deleting items by filter from {self.collection_prefix}_{collection_name}. Filter: {filter_string}'
|
||||
'Deleting items by filter from %s_%s. Filter: %s',
|
||||
self.collection_prefix,
|
||||
collection_name,
|
||||
filter_string,
|
||||
)
|
||||
return self.client.delete(
|
||||
collection_name=f'{self.collection_prefix}_{collection_name}',
|
||||
@@ -378,7 +391,7 @@ class MilvusClient(VectorDBBase):
|
||||
try:
|
||||
self.client.drop_collection(collection_name=collection_name_full)
|
||||
deleted_collections.append(collection_name_full)
|
||||
log.info(f'Deleted collection: {collection_name_full}')
|
||||
log.info('Deleted collection: %s', collection_name_full)
|
||||
except Exception as e:
|
||||
log.error(f'Error deleting collection {collection_name_full}: {e}')
|
||||
log.info(f'Milvus reset complete. Deleted collections: {deleted_collections}')
|
||||
log.info('Milvus reset complete. Deleted collections: %s', deleted_collections)
|
||||
@@ -146,7 +146,7 @@ class MilvusClient(VectorDBBase):
|
||||
# The index only accelerates resource_id filters; never fail
|
||||
# collection creation over it.
|
||||
log.warning(f'Could not create {RESOURCE_ID_FIELD} index on {mt_collection_name}: {e}')
|
||||
log.info(f'Created shared collection: {mt_collection_name}')
|
||||
log.info('Created shared collection: %s', mt_collection_name)
|
||||
|
||||
def _ensure_collection(self, mt_collection_name: str, dimension: int):
|
||||
if not self.client.has_collection(mt_collection_name):
|
||||
|
||||
@@ -181,7 +181,7 @@ class OpenGaussClient(VectorDBBase):
|
||||
new_items.append(new_chunk)
|
||||
self.session.bulk_save_objects(new_items)
|
||||
self.session.commit()
|
||||
log.info(f"Inserting {len(new_items)} items into collection '{collection_name}'.")
|
||||
log.info("Inserting %s items into collection '%s'.", len(new_items), collection_name)
|
||||
except Exception as e:
|
||||
self.session.rollback()
|
||||
log.exception(f'Failed to insert data: {e}')
|
||||
@@ -207,7 +207,7 @@ class OpenGaussClient(VectorDBBase):
|
||||
)
|
||||
self.session.add(new_chunk)
|
||||
self.session.commit()
|
||||
log.info(f"Inserting/updating {len(items)} items in collection '{collection_name}'.")
|
||||
log.info("Inserting/updating %s items in collection '%s'.", len(items), collection_name)
|
||||
except Exception as e:
|
||||
self.session.rollback()
|
||||
log.exception(f'Failed to insert or update data.: {e}')
|
||||
@@ -352,7 +352,7 @@ class OpenGaussClient(VectorDBBase):
|
||||
query = query.filter(DocumentChunk.vmetadata[key].astext == str(value))
|
||||
deleted = query.delete(synchronize_session=False)
|
||||
self.session.commit()
|
||||
log.info(f"Deleted {deleted} items from collection '{collection_name}'")
|
||||
log.info("Deleted %s items from collection '%s'", deleted, collection_name)
|
||||
except Exception as e:
|
||||
self.session.rollback()
|
||||
log.exception(f'Failed to delete data: {e}')
|
||||
@@ -362,7 +362,7 @@ class OpenGaussClient(VectorDBBase):
|
||||
try:
|
||||
deleted = self.session.query(DocumentChunk).delete()
|
||||
self.session.commit()
|
||||
log.info(f'Reset completed. Deleted {deleted} items')
|
||||
log.info('Reset completed. Deleted %s items', deleted)
|
||||
except Exception as e:
|
||||
self.session.rollback()
|
||||
log.exception(f'Reset failed: {e}')
|
||||
@@ -386,4 +386,4 @@ class OpenGaussClient(VectorDBBase):
|
||||
|
||||
def delete_collection(self, collection_name: str) -> None:
|
||||
self.delete(collection_name)
|
||||
log.info(f"Collection '{collection_name}' has been deleted")
|
||||
log.info("Collection '%s' has been deleted", collection_name)
|
||||
@@ -94,10 +94,10 @@ class Oracle23aiClient(VectorDBBase):
|
||||
self._create_dbcs_pool()
|
||||
|
||||
dsn = ORACLE_DB_DSN
|
||||
log.info(f'Creating Connection Pool [{ORACLE_DB_USER}:**@{dsn}]')
|
||||
log.info('Creating Connection Pool [%s:**@%s]', ORACLE_DB_USER, dsn)
|
||||
|
||||
with self.get_connection() as connection:
|
||||
log.info(f'Connection version: {connection.version}')
|
||||
log.info('Connection version: %s', connection.version)
|
||||
self._initialize_database(connection)
|
||||
|
||||
log.info('Oracle Vector Search initialization complete.')
|
||||
@@ -159,7 +159,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
|
||||
if attempt < max_retries - 1:
|
||||
wait_time = 2**attempt
|
||||
log.info(f'Retrying in {wait_time} seconds...')
|
||||
log.info('Retrying in %s seconds...', wait_time)
|
||||
time.sleep(wait_time)
|
||||
else:
|
||||
raise
|
||||
@@ -184,7 +184,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
|
||||
thread = threading.Thread(target=_monitor, daemon=True)
|
||||
thread.start()
|
||||
log.info(f'Started DB health monitor every {interval_seconds} seconds.')
|
||||
log.info('Started DB health monitor every %s seconds.', interval_seconds)
|
||||
|
||||
def _reconnect_pool(self):
|
||||
"""
|
||||
@@ -412,7 +412,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
... ]
|
||||
>>> client.insert("my_collection", items)
|
||||
"""
|
||||
log.info(f"Inserting {len(items)} items into collection '{collection_name}'.")
|
||||
log.info("Inserting %s items into collection '%s'.", len(items), collection_name)
|
||||
|
||||
with self.get_connection() as connection:
|
||||
try:
|
||||
@@ -437,7 +437,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
)
|
||||
|
||||
connection.commit()
|
||||
log.info(f"Successfully inserted {len(items)} items into collection '{collection_name}'.")
|
||||
log.info("Successfully inserted %s items into collection '%s'.", len(items), collection_name)
|
||||
|
||||
except Exception as e:
|
||||
connection.rollback()
|
||||
@@ -466,7 +466,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
... ]
|
||||
>>> client.upsert("my_collection", items)
|
||||
"""
|
||||
log.info(f"Upserting {len(items)} items into collection '{collection_name}'.")
|
||||
log.info("Upserting %s items into collection '%s'.", len(items), collection_name)
|
||||
|
||||
with self.get_connection() as connection:
|
||||
try:
|
||||
@@ -505,7 +505,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
)
|
||||
|
||||
connection.commit()
|
||||
log.info(f"Successfully upserted {len(items)} items into collection '{collection_name}'.")
|
||||
log.info("Successfully upserted %s items into collection '%s'.", len(items), collection_name)
|
||||
|
||||
except Exception as e:
|
||||
connection.rollback()
|
||||
@@ -541,7 +541,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
... for i, (id, dist) in enumerate(zip(results.ids[0], results.distances[0])):
|
||||
... log.info(f"Match {i+1}: id={id}, distance={dist}")
|
||||
"""
|
||||
log.info(f"Searching items from collection '{collection_name}' with limit {limit}.")
|
||||
log.info("Searching items from collection '%s' with limit %s.", collection_name, limit)
|
||||
|
||||
try:
|
||||
if not vectors:
|
||||
@@ -587,7 +587,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
metadatas[qid].append(self._json_to_metadata(metadata_str))
|
||||
distances[qid].append(float(row[3]))
|
||||
|
||||
log.info(f'Search completed. Found {sum(len(ids[i]) for i in range(num_queries))} total results.')
|
||||
log.info('Search completed. Found %s total results.', sum(len(ids[i]) for i in range(num_queries)))
|
||||
|
||||
return SearchResult(ids=ids, distances=distances, documents=documents, metadatas=metadatas)
|
||||
|
||||
@@ -616,7 +616,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
>>> if results:
|
||||
... print(f"Found {len(results.ids[0])} matching documents")
|
||||
"""
|
||||
log.info(f"Querying items from collection '{collection_name}' with filters.")
|
||||
log.info("Querying items from collection '%s' with filters.", collection_name)
|
||||
|
||||
try:
|
||||
limit = limit or 100
|
||||
@@ -656,7 +656,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
]
|
||||
]
|
||||
|
||||
log.info(f'Query completed. Found {len(results)} results.')
|
||||
log.info('Query completed. Found %s results.', len(results))
|
||||
|
||||
return GetResult(ids=ids, documents=documents, metadatas=metadatas)
|
||||
|
||||
@@ -747,7 +747,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
>>> # Or delete by metadata filter
|
||||
>>> client.delete("my_collection", filter={"source": "deprecated_source"})
|
||||
"""
|
||||
log.info(f"Deleting items from collection '{collection_name}'.")
|
||||
log.info("Deleting items from collection '%s'.", collection_name)
|
||||
|
||||
try:
|
||||
query = 'DELETE FROM document_chunk WHERE collection_name = :collection_name'
|
||||
@@ -772,7 +772,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
deleted = cursor.rowcount
|
||||
connection.commit()
|
||||
|
||||
log.info(f"Deleted {deleted} items from collection '{collection_name}'.")
|
||||
log.info("Deleted %s items from collection '%s'.", deleted, collection_name)
|
||||
|
||||
except Exception as e:
|
||||
log.exception(f'Error during delete: {e}')
|
||||
@@ -800,7 +800,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
deleted = cursor.rowcount
|
||||
connection.commit()
|
||||
|
||||
log.info(f"Reset complete. Deleted {deleted} items from 'document_chunk' table.")
|
||||
log.info("Reset complete. Deleted %s items from 'document_chunk' table.", deleted)
|
||||
|
||||
except Exception as e:
|
||||
log.exception(f'Error during reset: {e}')
|
||||
@@ -875,7 +875,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
>>> client = Oracle23aiClient()
|
||||
>>> client.delete_collection("obsolete_collection")
|
||||
"""
|
||||
log.info(f"Deleting collection '{collection_name}'.")
|
||||
log.info("Deleting collection '%s'.", collection_name)
|
||||
|
||||
try:
|
||||
with self.get_connection() as connection:
|
||||
@@ -891,7 +891,7 @@ class Oracle23aiClient(VectorDBBase):
|
||||
deleted = cursor.rowcount
|
||||
connection.commit()
|
||||
|
||||
log.info(f"Collection '{collection_name}' deleted. Removed {deleted} items.")
|
||||
log.info("Collection '%s' deleted. Removed %s items.", collection_name, deleted)
|
||||
|
||||
except Exception as e:
|
||||
log.exception(f"Error deleting collection '{collection_name}': {e}")
|
||||
|
||||
@@ -326,7 +326,7 @@ class PgvectorClient(VectorDBBase):
|
||||
},
|
||||
)
|
||||
self.session.commit()
|
||||
log.info(f"Encrypted & inserted {len(items)} into '{collection_name}'")
|
||||
log.info("Encrypted & inserted %s into '%s'", len(items), collection_name)
|
||||
|
||||
else:
|
||||
new_items = []
|
||||
@@ -342,7 +342,7 @@ class PgvectorClient(VectorDBBase):
|
||||
new_items.append(new_chunk)
|
||||
self.session.bulk_save_objects(new_items)
|
||||
self.session.commit()
|
||||
log.info(f"Inserted {len(new_items)} items into collection '{collection_name}'.")
|
||||
log.info("Inserted %s items into collection '%s'.", len(new_items), collection_name)
|
||||
except Exception as e:
|
||||
self.session.rollback()
|
||||
log.exception(f'Error during insert: {e}')
|
||||
@@ -381,7 +381,7 @@ class PgvectorClient(VectorDBBase):
|
||||
},
|
||||
)
|
||||
self.session.commit()
|
||||
log.info(f"Encrypted & upserted {len(items)} into '{collection_name}'")
|
||||
log.info("Encrypted & upserted %s into '%s'", len(items), collection_name)
|
||||
else:
|
||||
for item in items:
|
||||
vector = self.adjust_vector_length(item['vector'])
|
||||
@@ -401,7 +401,7 @@ class PgvectorClient(VectorDBBase):
|
||||
)
|
||||
self.session.add(new_chunk)
|
||||
self.session.commit()
|
||||
log.info(f"Upserted {len(items)} items into collection '{collection_name}'.")
|
||||
log.info("Upserted %s items into collection '%s'.", len(items), collection_name)
|
||||
except Exception as e:
|
||||
self.session.rollback()
|
||||
log.exception(f'Error during upsert: {e}')
|
||||
@@ -712,7 +712,7 @@ class PgvectorClient(VectorDBBase):
|
||||
query = query.filter(DocumentChunk.vmetadata[key].astext == str(value))
|
||||
deleted = query.delete(synchronize_session=False)
|
||||
self.session.commit()
|
||||
log.info(f"Deleted {deleted} items from collection '{collection_name}'.")
|
||||
log.info("Deleted %s items from collection '%s'.", deleted, collection_name)
|
||||
except Exception as e:
|
||||
self.session.rollback()
|
||||
log.exception(f'Error during delete: {e}')
|
||||
@@ -722,7 +722,7 @@ class PgvectorClient(VectorDBBase):
|
||||
try:
|
||||
deleted = self.session.query(DocumentChunk).delete()
|
||||
self.session.commit()
|
||||
log.info(f"Reset complete. Deleted {deleted} items from 'document_chunk' table.")
|
||||
log.info("Reset complete. Deleted %s items from 'document_chunk' table.", deleted)
|
||||
except Exception as e:
|
||||
self.session.rollback()
|
||||
log.exception(f'Error during reset: {e}')
|
||||
@@ -746,4 +746,4 @@ class PgvectorClient(VectorDBBase):
|
||||
|
||||
def delete_collection(self, collection_name: str) -> None:
|
||||
self.delete(collection_name)
|
||||
log.info(f"Collection '{collection_name}' deleted.")
|
||||
log.info("Collection '%s' deleted.", collection_name)
|
||||
@@ -106,16 +106,16 @@ class PineconeClient(VectorDBBase):
|
||||
try:
|
||||
# Check if index exists
|
||||
if self.index_name not in self.client.list_indexes().names():
|
||||
log.info(f"Creating Pinecone index '{self.index_name}'...")
|
||||
log.info("Creating Pinecone index '%s'...", self.index_name)
|
||||
self.client.create_index(
|
||||
name=self.index_name,
|
||||
dimension=self.dimension,
|
||||
metric=self.metric,
|
||||
spec=ServerlessSpec(cloud=self.cloud, region=self.environment),
|
||||
)
|
||||
log.info(f"Successfully created Pinecone index '{self.index_name}'")
|
||||
log.info("Successfully created Pinecone index '%s'", self.index_name)
|
||||
else:
|
||||
log.info(f"Using existing Pinecone index '{self.index_name}'")
|
||||
log.info("Using existing Pinecone index '%s'", self.index_name)
|
||||
|
||||
# Connect to the index
|
||||
self.index = self.client.Index(
|
||||
@@ -245,7 +245,7 @@ class PineconeClient(VectorDBBase):
|
||||
collection_name_with_prefix = self._get_collection_name_with_prefix(collection_name)
|
||||
try:
|
||||
self.index.delete(filter={'collection_name': collection_name_with_prefix})
|
||||
log.info(f"Collection '{collection_name_with_prefix}' deleted (all vectors removed).")
|
||||
log.info("Collection '%s' deleted (all vectors removed).", collection_name_with_prefix)
|
||||
except Exception as e:
|
||||
log.warning(f"Failed to delete collection '{collection_name_with_prefix}': {e}")
|
||||
raise
|
||||
@@ -276,7 +276,7 @@ class PineconeClient(VectorDBBase):
|
||||
elapsed = time.time() - start_time
|
||||
log.debug('Insert of %s vectors took %.2f seconds', len(points), elapsed)
|
||||
log.info(
|
||||
f"Successfully inserted {len(points)} vectors in parallel batches into '{collection_name_with_prefix}'"
|
||||
"Successfully inserted %s vectors in parallel batches into '%s'", len(points), collection_name_with_prefix
|
||||
)
|
||||
|
||||
def upsert(self, collection_name: str, items: List[VectorItem]) -> None:
|
||||
@@ -305,7 +305,7 @@ class PineconeClient(VectorDBBase):
|
||||
elapsed = time.time() - start_time
|
||||
log.debug('Upsert of %s vectors took %.2f seconds', len(points), elapsed)
|
||||
log.info(
|
||||
f"Successfully upserted {len(points)} vectors in parallel batches into '{collection_name_with_prefix}'"
|
||||
"Successfully upserted %s vectors in parallel batches into '%s'", len(points), collection_name_with_prefix
|
||||
)
|
||||
|
||||
async def insert_async(self, collection_name: str, items: List[VectorItem]) -> None:
|
||||
@@ -326,7 +326,9 @@ class PineconeClient(VectorDBBase):
|
||||
if isinstance(result, Exception):
|
||||
log.error(f'Error in async insert batch: {result}')
|
||||
raise result
|
||||
log.info(f"Successfully async inserted {len(points)} vectors in batches into '{collection_name_with_prefix}'")
|
||||
log.info(
|
||||
"Successfully async inserted %s vectors in batches into '%s'", len(points), collection_name_with_prefix
|
||||
)
|
||||
|
||||
async def upsert_async(self, collection_name: str, items: List[VectorItem]) -> None:
|
||||
"""Async version of upsert using asyncio and run_in_executor for improved performance."""
|
||||
@@ -346,7 +348,9 @@ class PineconeClient(VectorDBBase):
|
||||
if isinstance(result, Exception):
|
||||
log.error(f'Error in async upsert batch: {result}')
|
||||
raise result
|
||||
log.info(f"Successfully async upserted {len(points)} vectors in batches into '{collection_name_with_prefix}'")
|
||||
log.info(
|
||||
"Successfully async upserted %s vectors in batches into '%s'", len(points), collection_name_with_prefix
|
||||
)
|
||||
|
||||
def search(
|
||||
self,
|
||||
@@ -477,7 +481,7 @@ class PineconeClient(VectorDBBase):
|
||||
log.debug(
|
||||
"Deleted batch of %s vectors by ID from '%s'", len(batch_ids), collection_name_with_prefix
|
||||
)
|
||||
log.info(f"Successfully deleted {len(ids)} vectors by ID from '{collection_name_with_prefix}'")
|
||||
log.info("Successfully deleted %s vectors by ID from '%s'", len(ids), collection_name_with_prefix)
|
||||
|
||||
elif filter:
|
||||
# Combine user filter with collection_name
|
||||
@@ -486,7 +490,7 @@ class PineconeClient(VectorDBBase):
|
||||
pinecone_filter.update(filter)
|
||||
# Delete by metadata filter
|
||||
self.index.delete(filter=pinecone_filter)
|
||||
log.info(f"Successfully deleted vectors by filter from '{collection_name_with_prefix}'")
|
||||
log.info("Successfully deleted vectors by filter from '%s'", collection_name_with_prefix)
|
||||
|
||||
else:
|
||||
log.warning('No ids or filter provided for delete operation')
|
||||
|
||||
@@ -119,7 +119,7 @@ class QdrantClient(VectorDBBase):
|
||||
on_disk=self.QDRANT_ON_DISK,
|
||||
),
|
||||
)
|
||||
log.info(f'collection {collection_name_with_prefix} successfully created!')
|
||||
log.info('collection %s successfully created!', collection_name_with_prefix)
|
||||
|
||||
def _create_collection_if_not_exists(self, collection_name, dimension):
|
||||
if not self.has_collection(collection_name=collection_name):
|
||||
|
||||
@@ -148,7 +148,7 @@ class QdrantClient(VectorDBBase):
|
||||
m=0,
|
||||
),
|
||||
)
|
||||
log.info(f'Multi-tenant collection {mt_collection_name} created with dimension {dimension}!')
|
||||
log.info('Multi-tenant collection %s created with dimension %s!', mt_collection_name, dimension)
|
||||
|
||||
self.client.create_payload_index(
|
||||
collection_name=mt_collection_name,
|
||||
|
||||
@@ -36,7 +36,7 @@ class S3VectorClient(VectorDBBase):
|
||||
if self.bucket_name and self.region:
|
||||
try:
|
||||
self.client = boto3.client('s3vectors', region_name=self.region)
|
||||
log.info(f"S3Vector client initialized for bucket '{self.bucket_name}' in region '{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
|
||||
@@ -70,7 +70,9 @@ class S3VectorClient(VectorDBBase):
|
||||
]
|
||||
},
|
||||
)
|
||||
log.info(f'Created S3 index: {index_name} (dim={dimension}, type={data_type}, metric={distance_metric})')
|
||||
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
|
||||
@@ -137,9 +139,9 @@ class S3VectorClient(VectorDBBase):
|
||||
return
|
||||
|
||||
try:
|
||||
log.info(f"Deleting collection '{collection_name}'")
|
||||
log.info("Deleting collection '%s'", collection_name)
|
||||
self.client.delete_index(vectorBucketName=self.bucket_name, indexName=collection_name)
|
||||
log.info(f"Successfully deleted collection '{collection_name}'")
|
||||
log.info("Successfully deleted collection '%s'", collection_name)
|
||||
except Exception as e:
|
||||
log.error(f"Error deleting collection '{collection_name}': {e}")
|
||||
raise
|
||||
@@ -156,7 +158,7 @@ class S3VectorClient(VectorDBBase):
|
||||
|
||||
try:
|
||||
if not self.has_collection(collection_name):
|
||||
log.info(f"Index '{collection_name}' does not exist. Creating index.")
|
||||
log.info("Index '%s' does not exist. Creating index.", collection_name)
|
||||
self._create_index(
|
||||
index_name=collection_name,
|
||||
dimension=dimension,
|
||||
@@ -202,9 +204,11 @@ class S3VectorClient(VectorDBBase):
|
||||
indexName=collection_name,
|
||||
vectors=batch,
|
||||
)
|
||||
log.info(f"Inserted batch {i // batch_size + 1}: {len(batch)} vectors into index '{collection_name}'.")
|
||||
log.info(
|
||||
"Inserted batch %s: %s vectors into index '%s'.", i // batch_size + 1, len(batch), collection_name
|
||||
)
|
||||
|
||||
log.info(f"Completed insertion of {len(vectors)} vectors into index '{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
|
||||
@@ -218,11 +222,11 @@ class S3VectorClient(VectorDBBase):
|
||||
return
|
||||
|
||||
dimension = len(items[0]['vector'])
|
||||
log.info(f'Upsert dimension: {dimension}')
|
||||
log.info('Upsert dimension: %s', dimension)
|
||||
|
||||
try:
|
||||
if not self.has_collection(collection_name):
|
||||
log.info(f"Index '{collection_name}' does not exist. Creating index for upsert.")
|
||||
log.info("Index '%s' does not exist. Creating index for upsert.", collection_name)
|
||||
self._create_index(
|
||||
index_name=collection_name,
|
||||
dimension=dimension,
|
||||
@@ -264,10 +268,14 @@ class S3VectorClient(VectorDBBase):
|
||||
batch = vectors[i : i + batch_size]
|
||||
if i == 0: # Log sample info for first batch only
|
||||
log.info(
|
||||
f'Upserting batch 1: {len(batch)} vectors. First vector sample: key={batch[0]["key"]}, data_type={type(batch[0]["data"]["float32"])}, data_len={len(batch[0]["data"]["float32"])}'
|
||||
'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(f'Upserting batch {i // batch_size + 1}: {len(batch)} vectors.')
|
||||
log.info('Upserting batch %s: %s vectors.', i // batch_size + 1, len(batch))
|
||||
|
||||
self.client.put_vectors(
|
||||
vectorBucketName=self.bucket_name,
|
||||
@@ -275,7 +283,7 @@ class S3VectorClient(VectorDBBase):
|
||||
vectors=batch,
|
||||
)
|
||||
|
||||
log.info(f"Completed upsert of {len(vectors)} vectors into index '{collection_name}'.")
|
||||
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
|
||||
@@ -300,7 +308,7 @@ class S3VectorClient(VectorDBBase):
|
||||
return None
|
||||
|
||||
try:
|
||||
log.info(f"Searching collection '{collection_name}' with {len(vectors)} query vectors, limit={limit}")
|
||||
log.info("Searching collection '%s' with %s query vectors, limit=%s", collection_name, len(vectors), limit)
|
||||
|
||||
# Initialize result lists
|
||||
all_ids = []
|
||||
@@ -362,7 +370,7 @@ class S3VectorClient(VectorDBBase):
|
||||
all_metadatas.append(query_metadatas)
|
||||
all_distances.append(query_distances)
|
||||
|
||||
log.info(f'Search completed. Found results for {len(all_ids)} queries')
|
||||
log.info('Search completed. Found results for %s queries', len(all_ids))
|
||||
|
||||
# Return SearchResult format
|
||||
return SearchResult(
|
||||
@@ -402,7 +410,7 @@ class S3VectorClient(VectorDBBase):
|
||||
return self.get(collection_name)
|
||||
|
||||
try:
|
||||
log.info(f"Querying collection '{collection_name}' with filter: {filter}")
|
||||
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,
|
||||
@@ -437,7 +445,7 @@ class S3VectorClient(VectorDBBase):
|
||||
if limit and len(filtered_ids) >= limit:
|
||||
break
|
||||
|
||||
log.info(f'Filter applied: {len(filtered_ids)} vectors match out of {len(all_ids)} total')
|
||||
log.info('Filter applied: %s vectors match out of %s total', len(filtered_ids), len(all_ids))
|
||||
|
||||
# Return GetResult format
|
||||
if filtered_ids:
|
||||
@@ -472,7 +480,7 @@ class S3VectorClient(VectorDBBase):
|
||||
return GetResult(ids=[[]], documents=[[]], metadatas=[[]])
|
||||
|
||||
try:
|
||||
log.info(f"Retrieving all vectors from collection '{collection_name}'")
|
||||
log.info("Retrieving all vectors from collection '%s'", collection_name)
|
||||
|
||||
# Initialize result lists
|
||||
all_ids = []
|
||||
@@ -534,7 +542,7 @@ class S3VectorClient(VectorDBBase):
|
||||
if not next_token:
|
||||
break
|
||||
|
||||
log.info(f"Retrieved {len(all_ids)} vectors from collection '{collection_name}'")
|
||||
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
|
||||
@@ -576,17 +584,17 @@ class S3VectorClient(VectorDBBase):
|
||||
try:
|
||||
if ids:
|
||||
# Delete by specific vector IDs/keys
|
||||
log.info(f"Deleting {len(ids)} vectors by IDs from collection '{collection_name}'")
|
||||
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(f"Deleted {len(ids)} vectors from index '{collection_name}'")
|
||||
log.info("Deleted %s vectors from index '%s'", len(ids), collection_name)
|
||||
|
||||
elif filter:
|
||||
# Handle filter-based deletion
|
||||
log.info(f"Deleting vectors by filter from collection '{collection_name}': {filter}")
|
||||
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
|
||||
@@ -595,7 +603,8 @@ class S3VectorClient(VectorDBBase):
|
||||
file_collection_name = f'file-{file_id}'
|
||||
if self.has_collection(file_collection_name):
|
||||
log.info(
|
||||
f"Found related file-specific collection '{file_collection_name}', deleting it to prevent duplicates"
|
||||
"Found related file-specific collection '%s', deleting it to prevent duplicates",
|
||||
file_collection_name,
|
||||
)
|
||||
self.delete_collection(file_collection_name)
|
||||
|
||||
@@ -604,7 +613,7 @@ class S3VectorClient(VectorDBBase):
|
||||
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(f'Found {len(matching_ids)} vectors matching filter, deleting them')
|
||||
log.info('Found %s vectors matching filter, deleting them', len(matching_ids))
|
||||
|
||||
# Delete the matching vectors by ID
|
||||
self.client.delete_vectors(
|
||||
@@ -612,7 +621,7 @@ class S3VectorClient(VectorDBBase):
|
||||
indexName=collection_name,
|
||||
keys=matching_ids,
|
||||
)
|
||||
log.info(f"Deleted {len(matching_ids)} vectors from index '{collection_name}' using filter")
|
||||
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:
|
||||
@@ -645,11 +654,11 @@ class S3VectorClient(VectorDBBase):
|
||||
try:
|
||||
self.client.delete_index(vectorBucketName=self.bucket_name, indexName=index_name)
|
||||
deleted_count += 1
|
||||
log.info(f'Deleted index: {index_name}')
|
||||
log.info('Deleted index: %s', index_name)
|
||||
except Exception as e:
|
||||
log.error(f"Error deleting index '{index_name}': {e}")
|
||||
|
||||
log.info(f'Reset completed: deleted {deleted_count} indexes')
|
||||
log.info('Reset completed: deleted %s indexes', deleted_count)
|
||||
|
||||
except Exception as e:
|
||||
log.error(f'Error during reset: {e}')
|
||||
|
||||
@@ -279,7 +279,7 @@ class ValkeyClient(VectorDBBase):
|
||||
f'{self._format_version(MIN_VALKEY_VERSION)}. valkey-search 1.2.0 requires Valkey core '
|
||||
'9.0.1 or later. Upgrade your server or use valkey-bundle:9.1.0-rc2+.'
|
||||
)
|
||||
log.info(f'Valkey core version: {self._format_version(version) if version else "unknown"}')
|
||||
log.info('Valkey core version: %s', self._format_version(version) if version else 'unknown')
|
||||
|
||||
def _check_search_module(self) -> None:
|
||||
try:
|
||||
@@ -331,7 +331,7 @@ class ValkeyClient(VectorDBBase):
|
||||
'TEXT field type and filter-only FT.SEARCH support required by this backend. '
|
||||
'Upgrade to valkey-bundle:9.1.0-rc2+ or load valkey-search 1.2.0+ as a module.'
|
||||
)
|
||||
log.info(f'valkey-search version: {self._format_version(search_version) if search_version else "unknown"}')
|
||||
log.info('valkey-search version: %s', self._format_version(search_version) if search_version else 'unknown')
|
||||
|
||||
def _index_name(self, collection_name: str) -> str:
|
||||
return f'idx:{self.collection_prefix}:{collection_name}'
|
||||
@@ -385,8 +385,11 @@ class ValkeyClient(VectorDBBase):
|
||||
try:
|
||||
g['glide_ft'].create(self.client, index_name, schema, options)
|
||||
log.info(
|
||||
f'Created Valkey index {index_name} with dimension={dimension}, '
|
||||
f'type={self.index_type}, metric={self.distance_metric}'
|
||||
'Created Valkey index %s with dimension=%s, type=%s, metric=%s',
|
||||
index_name,
|
||||
dimension,
|
||||
self.index_type,
|
||||
self.distance_metric,
|
||||
)
|
||||
except g['RequestError'] as e:
|
||||
if 'already exists' in str(e).lower():
|
||||
@@ -456,7 +459,7 @@ class ValkeyClient(VectorDBBase):
|
||||
index_name = self._index_name(collection_name)
|
||||
try:
|
||||
self._g['glide_ft'].dropindex(self.client, index_name)
|
||||
log.info(f'Dropped index {index_name}')
|
||||
log.info('Dropped index %s', index_name)
|
||||
except self._g['RequestError'] as e:
|
||||
log.debug('Could not drop index %s: %s', index_name, e)
|
||||
|
||||
@@ -656,7 +659,7 @@ class ValkeyClient(VectorDBBase):
|
||||
collections.append(name[len(idx_prefix) :])
|
||||
try:
|
||||
glide_ft.dropindex(self.client, idx)
|
||||
log.info(f'Dropped index: {name}')
|
||||
log.info('Dropped index: %s', name)
|
||||
except Exception as e:
|
||||
log.error(f'Error dropping index {name}: {e}')
|
||||
except Exception as e:
|
||||
@@ -664,7 +667,7 @@ class ValkeyClient(VectorDBBase):
|
||||
|
||||
for collection in collections:
|
||||
self._delete_keys_by_prefix(self._key_prefix(collection))
|
||||
log.info(f'Valkey vector store reset complete (prefix: {self.collection_prefix})')
|
||||
log.info('Valkey vector store reset complete (prefix: %s)', self.collection_prefix)
|
||||
|
||||
def _delete_keys_by_prefix(self, prefix: str) -> None:
|
||||
cursor = '0'
|
||||
|
||||
@@ -31,7 +31,7 @@ def search_exa(
|
||||
count (int): Number of results to return
|
||||
filter_list (Optional[list[str]]): List of domains to filter results by
|
||||
"""
|
||||
log.info(f'Searching with Exa for query: {query}')
|
||||
log.info('Searching with Exa for query: %s', query)
|
||||
|
||||
headers = {'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'}
|
||||
|
||||
@@ -58,7 +58,7 @@ def search_exa(
|
||||
)
|
||||
)
|
||||
|
||||
log.info(f'Found {len(results)} results')
|
||||
log.info('Found %s results', len(results))
|
||||
return [
|
||||
SearchResult(
|
||||
link=result.url,
|
||||
|
||||
@@ -53,7 +53,7 @@ def search_external(
|
||||
)
|
||||
for result in results[:count]
|
||||
]
|
||||
log.info(f'External search results: {results}')
|
||||
log.info('External search results: %s', results)
|
||||
return results
|
||||
except Exception as e:
|
||||
log.error(f'Error in External search: {e}')
|
||||
|
||||
@@ -225,7 +225,7 @@ def search_firecrawl(
|
||||
)
|
||||
)
|
||||
|
||||
log.info(f'FireCrawl search results: {search_results}')
|
||||
log.info('FireCrawl search results: %s', search_results)
|
||||
return search_results
|
||||
except Exception as e:
|
||||
log.error(f'Error in FireCrawl search: {e}')
|
||||
|
||||
@@ -23,7 +23,7 @@ def search_ollama_cloud(
|
||||
count (int): Number of results to return
|
||||
filter_list (Optional[list[str]]): List of domains to filter results by
|
||||
"""
|
||||
log.info(f'Searching with Ollama for query: {query}')
|
||||
log.info('Searching with Ollama for query: %s', query)
|
||||
|
||||
headers = {'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'}
|
||||
payload = {'query': query, 'max_results': count}
|
||||
@@ -34,7 +34,7 @@ def search_ollama_cloud(
|
||||
data = response.json()
|
||||
|
||||
results = data.get('results', [])
|
||||
log.info(f'Found {len(results)} results')
|
||||
log.info('Found %s results', len(results))
|
||||
|
||||
if filter_list:
|
||||
results = get_filtered_results(results, filter_list)
|
||||
|
||||
@@ -31,7 +31,7 @@ def search_searchapi(
|
||||
response = requests.request('GET', url)
|
||||
|
||||
json_response = response.json()
|
||||
log.info(f'results from searchapi search: {json_response}')
|
||||
log.info('results from searchapi search: %s', json_response)
|
||||
|
||||
results = sorted(json_response.get('organic_results', []), key=lambda x: x.get('position', 0))
|
||||
if filter_list:
|
||||
|
||||
@@ -31,7 +31,7 @@ def search_serpapi(
|
||||
response = requests.request('GET', url)
|
||||
|
||||
json_response = response.json()
|
||||
log.info(f'results from serpapi search: {json_response}')
|
||||
log.info('results from serpapi search: %s', json_response)
|
||||
|
||||
results = sorted(json_response.get('organic_results', []), key=lambda x: x.get('position', 0))
|
||||
if filter_list:
|
||||
|
||||
@@ -51,7 +51,7 @@ def search_serply(
|
||||
response.raise_for_status()
|
||||
|
||||
json_response = response.json()
|
||||
log.info(f'results from serply search: {json_response}')
|
||||
log.info('results from serply search: %s', json_response)
|
||||
|
||||
results = sorted(json_response.get('results', []), key=lambda x: x.get('realPosition', 0))
|
||||
if filter_list:
|
||||
|
||||
@@ -112,7 +112,7 @@ def search_yandex(
|
||||
for result in results[:count]
|
||||
]
|
||||
|
||||
log.info(f'Yandex search results: {results}')
|
||||
log.info('Yandex search results: %s', results)
|
||||
|
||||
return results
|
||||
except Exception as e:
|
||||
|
||||
Reference in new issue
Block a user