mirror of
https://github.com/open-webui/open-webui.git
synced 2026-10-12 03:48:33 +00:00
8a90f0fc93
With Qdrant strict mode on and a max_query_limit below 999999999, Qdrant rejects Open WebUI's reads with "Limit exceeded", so every file upload after the first fails in the default multitenancy mode, and hybrid search finds nothing. Reads now go in pages of 1000 points, so any strict-mode limit of 1000 or more works. Without strict mode the results are the same as before. Tested against Qdrant 1.19.1 with max_query_limit 1000, for both multitenancy on and off: collections of up to 2500 points come back complete, limits are respected, and tenants stay separated. Fixes #31459
266 lines
10 KiB
Python
266 lines
10 KiB
Python
"""
|
|
NOTE: This vector database integration is community-supported and maintained on a best-effort basis.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any, Optional
|
|
from urllib.parse import urlparse
|
|
|
|
from open_webui.config import (
|
|
QDRANT_API_KEY,
|
|
QDRANT_COLLECTION_PREFIX,
|
|
QDRANT_GRPC_PORT,
|
|
QDRANT_HNSW_M,
|
|
QDRANT_ON_DISK,
|
|
QDRANT_PREFER_GRPC,
|
|
QDRANT_TIMEOUT,
|
|
QDRANT_URI,
|
|
)
|
|
from open_webui.retrieval.vector.main import (
|
|
GetResult,
|
|
SearchResult,
|
|
VectorDBBase,
|
|
VectorItem,
|
|
)
|
|
from open_webui.retrieval.vector.utils import iter_filter_conditions, process_metadata
|
|
from qdrant_client import QdrantClient as Qclient
|
|
from qdrant_client.http.models import PointStruct
|
|
from qdrant_client.models import models
|
|
|
|
NO_LIMIT = 999999999
|
|
SCROLL_PAGE_SIZE = 1000
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _metadata_filter(key: str, op: str, value: Any) -> models.FieldCondition:
|
|
match = models.MatchAny(any=value) if op == '$in' else models.MatchValue(value=value)
|
|
return models.FieldCondition(key=f'metadata.{key}', match=match)
|
|
|
|
|
|
class QdrantClient(VectorDBBase):
|
|
def __init__(self):
|
|
self.collection_prefix = QDRANT_COLLECTION_PREFIX
|
|
self.QDRANT_URI = QDRANT_URI
|
|
self.QDRANT_API_KEY = QDRANT_API_KEY
|
|
self.QDRANT_ON_DISK = QDRANT_ON_DISK
|
|
self.PREFER_GRPC = QDRANT_PREFER_GRPC
|
|
self.GRPC_PORT = QDRANT_GRPC_PORT
|
|
self.QDRANT_TIMEOUT = QDRANT_TIMEOUT
|
|
self.QDRANT_HNSW_M = QDRANT_HNSW_M
|
|
|
|
if not self.QDRANT_URI:
|
|
self.client = None
|
|
return
|
|
|
|
# Unified handling for either scheme
|
|
parsed = urlparse(self.QDRANT_URI)
|
|
host = parsed.hostname or self.QDRANT_URI
|
|
http_port = parsed.port or 6333 # default REST port
|
|
|
|
if self.PREFER_GRPC:
|
|
self.client = Qclient(
|
|
host=host,
|
|
port=http_port,
|
|
grpc_port=self.GRPC_PORT,
|
|
prefer_grpc=self.PREFER_GRPC,
|
|
api_key=self.QDRANT_API_KEY,
|
|
timeout=self.QDRANT_TIMEOUT,
|
|
)
|
|
else:
|
|
self.client = Qclient(
|
|
url=self.QDRANT_URI,
|
|
api_key=self.QDRANT_API_KEY,
|
|
timeout=QDRANT_TIMEOUT,
|
|
)
|
|
|
|
def _result_to_get_result(self, points) -> GetResult:
|
|
ids = []
|
|
documents = []
|
|
metadatas = []
|
|
|
|
for point in points:
|
|
payload = point.payload
|
|
ids.append(point.id)
|
|
documents.append(payload['text'])
|
|
metadatas.append(payload['metadata'])
|
|
|
|
return GetResult(
|
|
**{
|
|
'ids': [ids],
|
|
'documents': [documents],
|
|
'metadatas': [metadatas],
|
|
}
|
|
)
|
|
|
|
def _scroll_points(
|
|
self, collection_name: str, scroll_filter: Optional[models.Filter] = None, limit: Optional[int] = None
|
|
) -> list:
|
|
# Paged so a strict-mode max_query_limit does not reject the read
|
|
points = []
|
|
offset = None
|
|
while True:
|
|
page_size = SCROLL_PAGE_SIZE if limit is None else min(SCROLL_PAGE_SIZE, limit - len(points))
|
|
page, offset = self.client.scroll(
|
|
collection_name=f'{self.collection_prefix}_{collection_name}',
|
|
scroll_filter=scroll_filter,
|
|
limit=page_size,
|
|
offset=offset,
|
|
)
|
|
points.extend(page)
|
|
if offset is None or len(points) == limit:
|
|
return points
|
|
|
|
def _create_collection(self, collection_name: str, dimension: int):
|
|
collection_name_with_prefix = f'{self.collection_prefix}_{collection_name}'
|
|
self.client.create_collection(
|
|
collection_name=collection_name_with_prefix,
|
|
vectors_config=models.VectorParams(
|
|
size=dimension,
|
|
distance=models.Distance.COSINE,
|
|
on_disk=self.QDRANT_ON_DISK,
|
|
),
|
|
hnsw_config=models.HnswConfigDiff(
|
|
m=self.QDRANT_HNSW_M,
|
|
),
|
|
)
|
|
|
|
# Create payload indexes for efficient filtering
|
|
self.client.create_payload_index(
|
|
collection_name=collection_name_with_prefix,
|
|
field_name='metadata.hash',
|
|
field_schema=models.KeywordIndexParams(
|
|
type=models.KeywordIndexType.KEYWORD,
|
|
is_tenant=False,
|
|
on_disk=self.QDRANT_ON_DISK,
|
|
),
|
|
)
|
|
self.client.create_payload_index(
|
|
collection_name=collection_name_with_prefix,
|
|
field_name='metadata.file_id',
|
|
field_schema=models.KeywordIndexParams(
|
|
type=models.KeywordIndexType.KEYWORD,
|
|
is_tenant=False,
|
|
on_disk=self.QDRANT_ON_DISK,
|
|
),
|
|
)
|
|
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):
|
|
self._create_collection(collection_name=collection_name, dimension=dimension)
|
|
|
|
def _create_points(self, items: list[VectorItem]):
|
|
return [
|
|
PointStruct(
|
|
id=item['id'],
|
|
vector=item['vector'],
|
|
payload={'text': item['text'], 'metadata': process_metadata(item['metadata'])},
|
|
)
|
|
for item in items
|
|
]
|
|
|
|
def has_collection(self, collection_name: str) -> bool:
|
|
return self.client.collection_exists(f'{self.collection_prefix}_{collection_name}')
|
|
|
|
def delete_collection(self, collection_name: str):
|
|
return self.client.delete_collection(collection_name=f'{self.collection_prefix}_{collection_name}')
|
|
|
|
def search(
|
|
self,
|
|
collection_name: str,
|
|
vectors: list[list[float | int]],
|
|
filter: Optional[dict] = None,
|
|
limit: int = 10,
|
|
) -> Optional[SearchResult]:
|
|
# Search for the nearest neighbor items based on the vectors and return 'limit' number of results.
|
|
if limit is None:
|
|
limit = NO_LIMIT # otherwise qdrant would set limit to 10!
|
|
|
|
conditions = [_metadata_filter(key, op, value) for key, op, value in iter_filter_conditions(filter)]
|
|
query_filter = models.Filter(must=conditions) if conditions else None
|
|
query_response = self.client.query_points(
|
|
collection_name=f'{self.collection_prefix}_{collection_name}',
|
|
query=vectors[0],
|
|
limit=limit,
|
|
query_filter=query_filter,
|
|
)
|
|
get_result = self._result_to_get_result(query_response.points)
|
|
return SearchResult(
|
|
ids=get_result.ids,
|
|
documents=get_result.documents,
|
|
metadatas=get_result.metadatas,
|
|
# qdrant distance is [-1, 1], normalize to [0, 1]
|
|
distances=[[(point.score + 1.0) / 2.0 for point in query_response.points]],
|
|
)
|
|
|
|
def query(self, collection_name: str, filter: dict, limit: Optional[int] = None):
|
|
# Construct the filter string for querying
|
|
if not self.has_collection(collection_name):
|
|
return None
|
|
try:
|
|
field_conditions = []
|
|
for key, value in filter.items():
|
|
field_conditions.append(
|
|
models.FieldCondition(key=f'metadata.{key}', match=models.MatchValue(value=value))
|
|
)
|
|
|
|
points = self._scroll_points(collection_name, models.Filter(should=field_conditions), limit)
|
|
return self._result_to_get_result(points)
|
|
except Exception as e:
|
|
log.exception(f"Error querying a collection '{collection_name}': {e}")
|
|
return None
|
|
|
|
def get(self, collection_name: str) -> Optional[GetResult]:
|
|
# Get all the items in the collection.
|
|
points = self._scroll_points(collection_name)
|
|
return self._result_to_get_result(points)
|
|
|
|
def insert(self, collection_name: str, items: list[VectorItem]):
|
|
# Insert the items into the collection, if the collection does not exist, it will be created.
|
|
self._create_collection_if_not_exists(collection_name, len(items[0]['vector']))
|
|
points = self._create_points(items)
|
|
self.client.upload_points(f'{self.collection_prefix}_{collection_name}', points)
|
|
|
|
def upsert(self, collection_name: str, items: list[VectorItem]):
|
|
# Update the items in the collection, if the items are not present, insert them. If the collection does not exist, it will be created.
|
|
self._create_collection_if_not_exists(collection_name, len(items[0]['vector']))
|
|
points = self._create_points(items)
|
|
return self.client.upsert(f'{self.collection_prefix}_{collection_name}', points)
|
|
|
|
def delete(
|
|
self,
|
|
collection_name: str,
|
|
ids: Optional[list[str]] = None,
|
|
filter: Optional[dict] = None,
|
|
):
|
|
# Delete by point ID: the point ID is the item's id (see _create_points).
|
|
# Filtering on metadata.id silently misses points whose payload omits an
|
|
# id (e.g. memories), leaving orphaned vectors behind.
|
|
if ids:
|
|
return self.client.delete(
|
|
collection_name=f'{self.collection_prefix}_{collection_name}',
|
|
points_selector=models.PointIdsList(points=ids),
|
|
)
|
|
|
|
field_conditions = []
|
|
if filter:
|
|
for key, value in filter.items():
|
|
field_conditions.append(
|
|
models.FieldCondition(
|
|
key=f'metadata.{key}',
|
|
match=models.MatchValue(value=value),
|
|
)
|
|
)
|
|
|
|
return self.client.delete(
|
|
collection_name=f'{self.collection_prefix}_{collection_name}',
|
|
points_selector=models.FilterSelector(filter=models.Filter(must=field_conditions)),
|
|
)
|
|
|
|
def reset(self):
|
|
# Resets the database. This will delete all collections and item entries.
|
|
collection_names = self.client.get_collections().collections
|
|
for collection_name in collection_names:
|
|
if collection_name.name.startswith(self.collection_prefix):
|
|
self.client.delete_collection(collection_name=collection_name.name)
|